diff --git a/docs/CHEATSHEET.md b/docs/CHEATSHEET.md index 0dc77ba..ee8d9a7 100644 --- a/docs/CHEATSHEET.md +++ b/docs/CHEATSHEET.md @@ -341,12 +341,20 @@ No API credentials needed (uses public GQL endpoint). !alert check # Force-poll now ``` -Searches keywords across YouTube (InnerTube) and Twitch (GQL) simultaneously. +Searches keywords across YouTube (InnerTube), Twitch (GQL), and SearXNG simultaneously. Names: lowercase alphanumeric + hyphens, 1-20 chars. Keywords: 1-100 chars. Max 20 alerts/channel. Polls every 5min. Max 5 announcements per platform per cycle. -Format: `[name/yt] Title -- URL` or `[name/tw] Title -- URL`. +Format: `[name/yt] Title -- URL`, `[name/tw] Title -- URL`, or `[name/sx] Title -- URL`. No API credentials needed. Persists across restarts. +## SearX + +``` +!searx # Search SearXNG +``` + +Shows top 3 results as `Title -- URL`. Channel only. Max query length: 200 chars. + ## Plugin Template ```python diff --git a/docs/USAGE.md b/docs/USAGE.md index 1fbf4e3..be9204b 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -122,6 +122,7 @@ format = "text" # Log format: "text" (default) or "json" | `!username ` | Check username across ~25 services | | `!username ` | Check single service | | `!username list` | Show available services by category | +| `!searx ` | Search SearXNG and show top results | ### Command Shorthand @@ -630,11 +631,34 @@ Polling and announcements: - `list` shows live/error status indicators next to each streamer - `check` forces an immediate poll and reports current status +### `!searx` -- SearXNG Web Search + +Search the local SearXNG instance and display top results. + +``` +!searx Search SearXNG and show top results +``` + +- Open to all users, channel only (no PM) +- Query is everything after `!searx` +- Shows top 3 results as `Title -- URL` +- Titles truncated to 80 characters +- Query limited to 200 characters + +Output format: + +``` +Title One -- https://example.com/page1 +Title Two -- https://example.com/page2 +Title Three -- https://example.com/page3 +``` + ### `!alert` -- Keyword Alert Subscriptions -Search keywords across multiple platforms (YouTube, Twitch) and announce new -results. Unlike `!rss`/`!yt`/`!twitch` which follow specific channels/feeds, -`!alert` searches keywords across all supported platforms simultaneously. +Search keywords across multiple platforms (YouTube, Twitch, SearXNG) and +announce new results. Unlike `!rss`/`!yt`/`!twitch` which follow specific +channels/feeds, `!alert` searches keywords across all supported platforms +simultaneously. ``` !alert add Add keyword alert (admin) @@ -654,12 +678,14 @@ Platforms searched: - **YouTube** -- InnerTube search API (no auth required) - **Twitch** -- Public GQL endpoint: live streams and VODs (no auth required) +- **SearXNG** -- Local SearXNG instance (no auth required) Polling and announcements: - Alerts are polled every 5 minutes by default - On `add`, existing results are recorded without announcing (prevents flood) -- New results announced as `[name/yt] Title -- URL` or `[name/tw] Title -- URL` +- New results announced as `[name/yt] Title -- URL`, `[name/tw] Title -- URL`, + or `[name/sx] Title -- URL` - Maximum 5 items announced per platform per poll; excess shown as `... and N more` - Titles are truncated to 80 characters - Each platform maintains its own seen list (capped at 200 per platform) diff --git a/plugins/alert.py b/plugins/alert.py index ffbb39b..3c443f4 100644 --- a/plugins/alert.py +++ b/plugins/alert.py @@ -26,6 +26,7 @@ _YT_SEARCH_URL = "https://www.youtube.com/youtubei/v1/search" _YT_CLIENT_VERSION = "2.20250101.00.00" _GQL_URL = "https://gql.twitch.tv/gql" _GQL_CLIENT_ID = "kimne78kx3ncx6brgo4mv6wki5h1ko" +_SEARX_URL = "http://192.168.122.119:3000/search" # -- Module-level tracking --------------------------------------------------- @@ -193,11 +194,40 @@ def _search_twitch(keyword: str) -> list[dict]: return results +# -- SearXNG search (blocking) ---------------------------------------------- + +def _search_searx(keyword: str) -> list[dict]: + """Search SearXNG. Blocking.""" + import urllib.parse + + params = urllib.parse.urlencode({"q": keyword, "format": "json"}) + url = f"{_SEARX_URL}?{params}" + + req = urllib.request.Request(url, method="GET") + resp = urllib.request.urlopen(req, timeout=_FETCH_TIMEOUT) + raw = resp.read() + resp.close() + + data = json.loads(raw) + results: list[dict] = [] + for item in data.get("results", []): + item_url = item.get("url", "") + title = item.get("title", "") + results.append({ + "id": item_url, + "title": title, + "url": item_url, + "extra": "", + }) + return results + + # -- Backend registry ------------------------------------------------------- _BACKENDS: dict[str, callable] = { "yt": _search_youtube, "tw": _search_twitch, + "sx": _search_searx, } diff --git a/plugins/searx.py b/plugins/searx.py new file mode 100644 index 0000000..31f131a --- /dev/null +++ b/plugins/searx.py @@ -0,0 +1,94 @@ +"""Plugin: SearXNG web search.""" + +from __future__ import annotations + +import json +import urllib.parse +import urllib.request + +from derp.plugin import command + +# -- Constants --------------------------------------------------------------- + +_SEARX_URL = "http://192.168.122.119:3000/search" +_FETCH_TIMEOUT = 10 +_MAX_RESULTS = 3 +_MAX_TITLE_LEN = 80 +_MAX_QUERY_LEN = 200 + + +# -- Pure helpers ------------------------------------------------------------ + +def _truncate(text: str, max_len: int = _MAX_TITLE_LEN) -> str: + """Truncate text with ellipsis if needed.""" + if len(text) <= max_len: + return text + return text[: max_len - 3].rstrip() + "..." + + +# -- SearXNG search (blocking) ---------------------------------------------- + +def _search(query: str) -> list[dict]: + """Search SearXNG. Blocking. + + Returns list of dicts with keys: title, url, snippet. + Raises on HTTP or parse errors. + """ + params = urllib.parse.urlencode({"q": query, "format": "json"}) + url = f"{_SEARX_URL}?{params}" + + req = urllib.request.Request(url, method="GET") + resp = urllib.request.urlopen(req, timeout=_FETCH_TIMEOUT) + raw = resp.read() + resp.close() + + data = json.loads(raw) + results: list[dict] = [] + for item in data.get("results", []): + results.append({ + "title": item.get("title", ""), + "url": item.get("url", ""), + "snippet": item.get("content", item.get("snippet", "")), + }) + return results + + +# -- Command handler --------------------------------------------------------- + +@command("searx", help="Search: !searx ") +async def cmd_searx(bot, message): + """Search SearXNG and show top results. + + Usage: !searx + """ + if not message.is_channel: + await bot.reply(message, "Use this command in a channel") + return + + parts = message.text.split(None, 1) + if len(parts) < 2 or not parts[1].strip(): + await bot.reply(message, "Usage: !searx ") + return + + query = parts[1].strip() + if len(query) > _MAX_QUERY_LEN: + await bot.reply(message, f"Query too long (max {_MAX_QUERY_LEN} chars)") + return + + import asyncio + loop = asyncio.get_running_loop() + + try: + results = await loop.run_in_executor(None, _search, query) + except Exception as exc: + await bot.reply(message, f"Search failed: {exc}") + return + + if not results: + await bot.reply(message, f"No results for: {query}") + return + + for item in results[:_MAX_RESULTS]: + title = _truncate(item["title"]) if item["title"] else "(no title)" + url = item["url"] + await bot.reply(message, f"{title} -- {url}") diff --git a/tests/test_alert.py b/tests/test_alert.py index 6903c36..e9885a8 100644 --- a/tests/test_alert.py +++ b/tests/test_alert.py @@ -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") diff --git a/tests/test_searx.py b/tests/test_searx.py new file mode 100644 index 0000000..e092402 --- /dev/null +++ b/tests/test_searx.py @@ -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())