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

fix: repair broken tests across alert, chanmgmt, and integration

Browse Source 
- test_alert: remove stale _MAX_ANNOUNCE import/test, update _errors
  assertions for per-backend dict, fix announcement checks (action vs
  send), mock _fetch_og_batch in seeding tests, fix YouTube/SearX mock
  targets (urllib.request.urlopen), include keyword in fake data titles
- test_chanmgmt: add _FakeState to _FakeBot (on_invite now persists)
- test_integration: update help assertion for new output format

696 tests pass, 0 failures.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
user
2026-02-17 21:14:44 +01:00
parent 94f563d55a
commit 7606280358
3 changed files with 114 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

190
tests/test_alert.py
View File

@@ -18,7 +18,6 @@ sys.modules[_spec.name] = _mod
_spec.loader.exec_module(_mod)
from plugins.alert import ( # noqa: E402
_MAX_ANNOUNCE,
_MAX_SEEN,
_delete,
_errors,
@@ -153,18 +152,30 @@ class _FakeState:
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.actions: list[tuple[str, str]] = []
self.replied: list[str] = []
self.state = _FakeState()
self.registry = _FakeRegistry()
self._admin = admin
async def send(self, target: str, text: str) -> None:
self.sent.append((target, text))
async def action(self, target: str, text: str) -> None:
self.actions.append((target, text))
async def reply(self, message, text: str) -> None:
self.replied.append(text)
@@ -199,21 +210,21 @@ def _clear() -> None:
def _fake_yt(keyword):
"""Fake YouTube backend returning two results."""
"""Fake YouTube backend returning two results (keyword in title)."""
return [
{"id": "yt1", "title": "YT Result 1",
{"id": "yt1", "title": "YT test Result 1",
"url": "https://www.youtube.com/watch?v=yt1", "extra": ""},
{"id": "yt2", "title": "YT Result 2",
{"id": "yt2", "title": "YT test Result 2",
"url": "https://www.youtube.com/watch?v=yt2", "extra": ""},
]
def _fake_tw(keyword):
"""Fake Twitch backend returning two results."""
"""Fake Twitch backend returning two results (keyword in title)."""
return [
{"id": "stream:tw1", "title": "TW Stream 1",
{"id": "stream:tw1", "title": "TW test Stream 1",
"url": "https://twitch.tv/user1", "extra": ""},
{"id": "vod:tw2", "title": "TW VOD 1",
{"id": "vod:tw2", "title": "TW test VOD 1",
"url": "https://twitch.tv/videos/tw2", "extra": ""},
]
@@ -229,11 +240,11 @@ def _fake_tw_error(keyword):
def _fake_sx(keyword):
"""Fake SearX backend returning two results."""
"""Fake SearX backend returning two results (keyword in title)."""
return [
{"id": "https://example.com/sx1", "title": "SX Result 1",
{"id": "https://example.com/sx1", "title": "SX test Result 1",
"url": "https://example.com/sx1", "extra": ""},
{"id": "https://example.com/sx2", "title": "SX Result 2",
{"id": "https://example.com/sx2", "title": "SX test Result 2",
"url": "https://example.com/sx2", "extra": ""},
]
@@ -370,7 +381,7 @@ class TestExtractVideos:
def close(self):
pass
with patch.object(_mod, "_urlopen", return_value=FakeResp()):
with patch("urllib.request.urlopen", return_value=FakeResp()):
results = _search_youtube("test")
assert len(results) == 1
assert results[0]["id"] == "dup1"
@@ -388,7 +399,7 @@ class TestSearchYoutube:
def close(self):
pass
with patch.object(_mod, "_urlopen", return_value=FakeResp()):
with patch("urllib.request.urlopen", return_value=FakeResp()):
results = _search_youtube("test query")
assert len(results) == 2
assert results[0]["id"] == "abc123"
@@ -396,7 +407,7 @@ class TestSearchYoutube:
def test_http_error_propagates(self):
import pytest
with patch.object(_mod, "_urlopen", side_effect=ConnectionError("fail")):
with patch("urllib.request.urlopen", side_effect=ConnectionError("fail")):
with pytest.raises(ConnectionError):
_search_youtube("test")
@@ -529,26 +540,28 @@ class TestCmdAlertAdd:
bot = _FakeBot(admin=True)
async def inner():
with patch.object(_mod, "_BACKENDS", _FAKE_BACKENDS):
with (
patch.object(_mod, "_BACKENDS", _FAKE_BACKENDS),
patch.object(_mod, "_fetch_og_batch", return_value={}),
):
await cmd_alert(bot, _msg("!alert add mc-speed minecraft speedrun"))
await asyncio.sleep(0)
assert len(bot.replied) == 1
assert "Alert 'mc-speed' added" in bot.replied[0]
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"
assert data["keyword"] == "minecraft speedrun"
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)
# Allow background seeding task to complete (patches must stay active)
await asyncio.sleep(0.2)
assert len(bot.replied) == 1
assert "Alert 'mc-speed' added" in bot.replied[0]
assert "minecraft speedrun" in bot.replied[0]
data = _load(bot, "#test:mc-speed")
assert data is not None
assert data["name"] == "mc-speed"
assert data["keyword"] == "minecraft speedrun"
assert data["channel"] == "#test"
# Seeding happens in background; verify seen lists populated
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)
asyncio.run(inner())
@@ -590,7 +603,7 @@ class TestCmdAlertAdd:
async def inner():
with patch.object(_mod, "_BACKENDS", _FAKE_BACKENDS):
await cmd_alert(bot, _msg("!alert add dupe some keyword"))
await asyncio.sleep(0)
await asyncio.sleep(0.1)
bot.replied.clear()
with patch.object(_mod, "_BACKENDS", _FAKE_BACKENDS):
await cmd_alert(bot, _msg("!alert add dupe other keyword"))
@@ -620,16 +633,20 @@ class TestCmdAlertAdd:
backends = {"yt": _fake_yt, "tw": _fake_tw_error, "sx": _fake_sx}
async def inner():
with patch.object(_mod, "_BACKENDS", backends):
with (
patch.object(_mod, "_BACKENDS", backends),
patch.object(_mod, "_fetch_og_batch", return_value={}),
):
await cmd_alert(bot, _msg("!alert add partial test keyword"))
await asyncio.sleep(0)
data = _load(bot, "#test:partial")
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)
# Allow background seeding task to complete (patches must stay active)
await asyncio.sleep(0.2)
data = _load(bot, "#test:partial")
assert data is not None
assert len(data["seen"]["yt"]) == 2
assert len(data["seen"].get("tw", [])) == 0
assert len(data["seen"]["sx"]) == 2
_stop_poller("#test:partial")
await asyncio.sleep(0)
asyncio.run(inner())
@@ -646,7 +663,7 @@ class TestCmdAlertDel:
async def inner():
with patch.object(_mod, "_BACKENDS", _FAKE_BACKENDS):
await cmd_alert(bot, _msg("!alert add todel some keyword"))
await asyncio.sleep(0)
await asyncio.sleep(0.1)
bot.replied.clear()
await cmd_alert(bot, _msg("!alert del todel"))
assert "Removed 'todel'" in bot.replied[0]
@@ -713,10 +730,11 @@ class TestCmdAlertList:
bot = _FakeBot()
_save(bot, "#test:broken", {
"name": "broken", "channel": "#test", "keyword": "test",
"last_error": "Connection refused",
"last_errors": {"yt": "Connection refused"},
})
asyncio.run(cmd_alert(bot, _msg("!alert list")))
assert "broken (error)" in bot.replied[0]
assert "broken" in bot.replied[0]
assert "backend error" in bot.replied[0]
def test_list_requires_channel(self):
_clear()
@@ -809,10 +827,11 @@ class TestCmdAlertCheck:
with patch.object(_mod, "_BACKENDS", _FAKE_BACKENDS):
await cmd_alert(bot, _msg("!alert check news"))
# 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]
# Metadata (with backend tags) goes to action(), titles to send()
actions = [s for t, s in bot.actions if t == "#test"]
yt_msgs = [m for m in actions if "/yt/" in m]
tw_msgs = [m for m in actions if "/tw/" in m]
sx_msgs = [m for m in actions 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
@@ -846,11 +865,14 @@ class TestPollOnce:
async def inner():
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) == 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]
# Titles go to send(), metadata goes to action()
titles = [s for t, s in bot.sent if t == "#test"]
actions = [s for t, s in bot.actions if t == "#test"]
assert len(titles) == 6 # 2 yt + 2 tw + 2 sx
assert len(actions) == 6
assert "[poll/yt/" in actions[0]
assert "[poll/tw/" in actions[2]
assert "[poll/sx/" in actions[4]
asyncio.run(inner())
@@ -877,36 +899,6 @@ class TestPollOnce:
asyncio.run(inner())
def test_max_announce_per_platform(self):
"""Only MAX_ANNOUNCE items per platform, then '... and N more'."""
_clear()
bot = _FakeBot()
def fake_many(keyword):
return [
{"id": f"v{i}", "title": f"Video {i}",
"url": f"https://example.com/{i}", "extra": ""}
for i in range(8)
]
data = {
"keyword": "test", "name": "many", "channel": "#test",
"interval": 300, "seen": {"yt": [], "tw": [], "sx": []},
"last_poll": "", "last_error": "",
}
key = "#test:many"
_save(bot, key, data)
_subscriptions[key] = data
async def inner():
with patch.object(_mod, "_BACKENDS", {"yt": fake_many, "tw": _fake_tw}):
await _poll_once(bot, key, announce=True)
yt_msgs = [s for t, s in bot.sent if t == "#test" and "/yt]" in s]
assert len(yt_msgs) == _MAX_ANNOUNCE + 1 # 5 items + "... and 3 more"
assert "... and 3 more" in yt_msgs[-1]
asyncio.run(inner())
def test_partial_backend_failure(self):
"""One backend fails, other still works. Error counter increments."""
_clear()
@@ -925,14 +917,14 @@ class TestPollOnce:
with patch.object(_mod, "_BACKENDS", backends):
await _poll_once(bot, key, announce=True)
# 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]
tw_msgs = [s for t, s in bot.actions if t == "#test" and "/tw/" in s]
sx_msgs = [s for t, s in bot.actions 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
# Error counter should be incremented for yt backend
assert _errors[key]["yt"] == 1
updated = _load(bot, key)
assert "yt:" in updated["last_error"]
assert "yt" in updated.get("last_errors", {})
asyncio.run(inner())
@@ -1005,7 +997,7 @@ class TestPollOnce:
async def inner():
with patch.object(_mod, "_BACKENDS", backends):
await _poll_once(bot, key, announce=True)
assert _errors[key] == 1
assert all(v == 1 for v in _errors[key].values())
assert len(bot.sent) == 0
asyncio.run(inner())
@@ -1019,19 +1011,19 @@ class TestPollOnce:
"yt": ["yt1", "yt2"], "tw": ["stream:tw1", "vod:tw2"],
"sx": ["https://example.com/sx1", "https://example.com/sx2"],
},
"last_poll": "", "last_error": "old error",
"last_poll": "", "last_errors": {"yt": "old error"},
}
key = "#test:clrerr"
_save(bot, key, data)
_subscriptions[key] = data
_errors[key] = 3
_errors[key] = {"yt": 3, "tw": 3, "sx": 3}
async def inner():
with patch.object(_mod, "_BACKENDS", _FAKE_BACKENDS):
await _poll_once(bot, key, announce=True)
assert _errors[key] == 0
assert all(v == 0 for v in _errors[key].values())
updated = _load(bot, key)
assert updated["last_error"] == ""
assert updated.get("last_errors", {}) == {}
asyncio.run(inner())
@@ -1222,6 +1214,7 @@ class TestSearchSearx:
with patch("urllib.request.urlopen", return_value=FakeResp()):
results = _search_searx("test query")
# Same response served for all categories; deduped by URL
assert len(results) == 3
assert results[0]["id"] == "https://example.com/sx1"
assert results[0]["title"] == "SearX Result 1"
@@ -1241,9 +1234,8 @@ class TestSearchSearx:
results = _search_searx("nothing")
assert results == []
def test_http_error_propagates(self):
import pytest
def test_http_error_returns_empty(self):
"""SearXNG catches per-category errors; all failing returns empty."""
with patch("urllib.request.urlopen", side_effect=ConnectionError("fail")):
with pytest.raises(ConnectionError):
_search_searx("test")
results = _search_searx("test")
assert results == []

21
tests/test_chanmgmt.py
View File

@@ -31,6 +31,26 @@ class _FakeConn:
self.sent.append(raw)
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
class _FakeBot:
"""Minimal bot stand-in."""
@@ -38,6 +58,7 @@ class _FakeBot:
self.joined: list[str] = []
self._admin = admin
self.conn = _FakeConn()
self.state = _FakeState()
def _is_admin(self, message) -> bool:
return self._admin

4
tests/test_integration.py
View File

@@ -227,8 +227,8 @@ class TestCommandDispatch:
replies = h.sent_privmsgs("#test")
assert len(replies) == 1
assert "Commands:" in replies[0]
assert "!ping" in replies[0]
assert "help" in replies[0]
assert "ping" in replies[0]
def test_unknown_command_ignored(self):
"""Unknown commands produce no reply."""