username/bouncer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
bfcebad6dda4d269aef676f8586b6714db2c3f13
bouncer/tests/test_router.py
user 0d762ced49 feat: PING watchdog, IRCv3 server-time, push notifications 
PING watchdog sends PING after configurable silence interval and
disconnects on timeout, detecting stale connections that TCP alone
misses. IRCv3 server-time capability is requested on every connection;
timestamps are injected on dispatch and backlog replay for clients
that support message tags. Push notifications via ntfy or generic
webhook fire on highlights and PMs when no clients are attached,
with configurable cooldown and optional SOCKS5 routing.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-21 17:41:38 +01:00

933 lines
32 KiB
Python
Raw Blame History

"""Tests for message router."""
from __future__ import annotations
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from bouncer.config import BouncerConfig, Config, NetworkConfig, ProxyConfig
from bouncer.irc import IRCMessage
from bouncer.router import BACKLOG_COMMANDS, Router, _suppress
# -- helpers -----------------------------------------------------------------
def _net_cfg(name: str = "libera", host: str = "irc.libera.chat",
port: int = 6697, tls: bool = True,
proxy_host: str | None = None,
proxy_port: int | None = None) -> NetworkConfig:
return NetworkConfig(
name=name, host=host, port=port, tls=tls,
proxy_host=proxy_host, proxy_port=proxy_port,
)
def _config(*net_cfgs: NetworkConfig) -> Config:
nets = {n.name: n for n in net_cfgs} if net_cfgs else {
"libera": _net_cfg("libera"),
}
return Config(
bouncer=BouncerConfig(),
proxy=ProxyConfig(host="127.0.0.1", port=1080),
networks=nets,
)
def _backlog() -> AsyncMock:
bl = AsyncMock()
bl.get_last_seen = AsyncMock(return_value=0)
bl.replay = AsyncMock(return_value=[])
bl.mark_seen = AsyncMock()
return bl
def _mock_network(name: str = "libera", nick: str = "botnick",
connected: bool = True, channels: set[str] | None = None,
topics: dict[str, str] | None = None,
names: dict[str, set[str]] | None = None) -> MagicMock:
net = MagicMock()
net.cfg.name = name
net.nick = nick
net.connected = connected
net.channels = channels or set()
net.topics = topics or {}
net.names = names or {}
net.send = AsyncMock()
net.stop = AsyncMock()
net.start = AsyncMock()
return net
def _mock_client(nick: str = "testuser") -> MagicMock:
client = MagicMock()
client.nick = nick
client.write = MagicMock()
return client
def _msg(command: str, params: list[str] | None = None,
prefix: str | None = None, tags: dict | None = None) -> IRCMessage:
return IRCMessage(
command=command,
params=params or [],
prefix=prefix,
tags=tags or {},
)
# -- _suppress ---------------------------------------------------------------
class TestSuppress:
def test_suppresses_welcome_numerics(self) -> None:
for num in ("001", "002", "003", "004", "005"):
assert _suppress(_msg(num, ["nick", "text"])) is True
def test_suppresses_motd(self) -> None:
for num in ("375", "372", "376", "422"):
assert _suppress(_msg(num, ["nick", "text"])) is True
def test_suppresses_lusers(self) -> None:
for num in ("250", "251", "252", "253", "254", "255", "265", "266"):
assert _suppress(_msg(num, ["nick", "text"])) is True
def test_suppresses_uid_and_visiblehost(self) -> None:
assert _suppress(_msg("042", ["nick", "UID"])) is True
assert _suppress(_msg("396", ["nick", "host"])) is True
def test_suppresses_nick_in_use(self) -> None:
assert _suppress(_msg("433", ["*", "nick", "in use"])) is True
def test_suppresses_server_notice(self) -> None:
msg = _msg("NOTICE", ["nick", "server message"], prefix="server.example.com")
assert _suppress(msg) is True
def test_passes_user_notice(self) -> None:
msg = _msg("NOTICE", ["nick", "hello"], prefix="user!ident@host")
assert _suppress(msg) is False
def test_suppresses_connection_notice_star(self) -> None:
msg = _msg("NOTICE", ["*", "Looking up your hostname..."])
assert _suppress(msg) is True
def test_suppresses_connection_notice_auth(self) -> None:
msg = _msg("NOTICE", ["AUTH", "*** Checking Ident"])
assert _suppress(msg) is True
def test_suppresses_ctcp_reply_in_notice(self) -> None:
msg = _msg("NOTICE", ["nick", "\x01VERSION mIRC\x01"], prefix="user!i@h")
assert _suppress(msg) is True
def test_suppresses_ctcp_in_privmsg(self) -> None:
msg = _msg("PRIVMSG", ["nick", "\x01VERSION\x01"], prefix="user!i@h")
assert _suppress(msg) is True
def test_passes_action_in_privmsg(self) -> None:
msg = _msg("PRIVMSG", ["#ch", "\x01ACTION waves\x01"], prefix="user!i@h")
assert _suppress(msg) is False
def test_suppresses_user_mode(self) -> None:
msg = _msg("MODE", ["nick", "+i"])
assert _suppress(msg) is True
def test_passes_channel_mode(self) -> None:
msg = _msg("MODE", ["#channel", "+o", "nick"])
assert _suppress(msg) is False
def test_passes_normal_privmsg(self) -> None:
msg = _msg("PRIVMSG", ["#test", "hello"], prefix="user!i@h")
assert _suppress(msg) is False
def test_passes_join(self) -> None:
msg = _msg("JOIN", ["#test"], prefix="user!i@h")
assert _suppress(msg) is False
def test_passes_part(self) -> None:
msg = _msg("PART", ["#test"], prefix="user!i@h")
assert _suppress(msg) is False
def test_passes_kick(self) -> None:
msg = _msg("KICK", ["#test", "nick", "reason"], prefix="op!i@h")
assert _suppress(msg) is False
def test_passes_topic(self) -> None:
msg = _msg("TOPIC", ["#test", "new topic"], prefix="user!i@h")
assert _suppress(msg) is False
# -- Router._proxy_for ------------------------------------------------------
class TestProxyFor:
def test_default_proxy(self) -> None:
cfg = _config()
router = Router(cfg, _backlog())
proxy = router._proxy_for(_net_cfg())
assert proxy.host == "127.0.0.1"
assert proxy.port == 1080
def test_per_network_proxy_override(self) -> None:
cfg = _config()
router = Router(cfg, _backlog())
net = _net_cfg(proxy_host="10.0.0.1", proxy_port=9050)
proxy = router._proxy_for(net)
assert proxy.host == "10.0.0.1"
assert proxy.port == 9050
def test_per_network_proxy_inherits_port(self) -> None:
cfg = _config()
router = Router(cfg, _backlog())
net = _net_cfg(proxy_host="10.0.0.1")
proxy = router._proxy_for(net)
assert proxy.host == "10.0.0.1"
assert proxy.port == 1080 # from global config
# -- Router init and network management --------------------------------------
class TestNetworkManagement:
def test_network_names_empty(self) -> None:
cfg = _config()
router = Router(cfg, _backlog())
assert router.network_names() == []
def test_get_network_none(self) -> None:
cfg = _config()
router = Router(cfg, _backlog())
assert router.get_network("nonexistent") is None
def test_get_network_found(self) -> None:
cfg = _config()
router = Router(cfg, _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
assert router.get_network("libera") is net
def test_network_names(self) -> None:
cfg = _config()
router = Router(cfg, _backlog())
router.networks["libera"] = _mock_network("libera")
router.networks["oftc"] = _mock_network("oftc")
assert sorted(router.network_names()) == ["libera", "oftc"]
def test_get_own_nicks(self) -> None:
cfg = _config()
router = Router(cfg, _backlog())
router.networks["libera"] = _mock_network("libera", nick="lnick")
router.networks["oftc"] = _mock_network("oftc", nick="onick")
nicks = router.get_own_nicks()
assert nicks == {"libera": "lnick", "oftc": "onick"}
@pytest.mark.asyncio
async def test_add_network(self) -> None:
cfg = _config()
router = Router(cfg, _backlog())
net_cfg = _net_cfg("hackint", host="irc.hackint.org")
with patch("bouncer.router.Network") as MockNet:
mock_instance = MagicMock()
mock_instance.start = AsyncMock()
MockNet.return_value = mock_instance
result = await router.add_network(net_cfg)
assert "hackint" in router.networks
assert result is mock_instance
@pytest.mark.asyncio
async def test_remove_network(self) -> None:
cfg = _config()
router = Router(cfg, _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
result = await router.remove_network("libera")
assert result is True
assert "libera" not in router.networks
net.stop.assert_awaited_once()
@pytest.mark.asyncio
async def test_remove_network_not_found(self) -> None:
cfg = _config()
router = Router(cfg, _backlog())
result = await router.remove_network("nonexistent")
assert result is False
# -- Client attach/detach ---------------------------------------------------
class TestClientAttachDetach:
@pytest.mark.asyncio
async def test_attach_adds_client(self) -> None:
router = Router(_config(), _backlog())
client = _mock_client()
await router.attach_all(client)
assert client in router.clients
@pytest.mark.asyncio
async def test_detach_removes_client(self) -> None:
router = Router(_config(), _backlog())
client = _mock_client()
await router.attach_all(client)
await router.detach_all(client)
assert client not in router.clients
@pytest.mark.asyncio
async def test_detach_missing_client_no_error(self) -> None:
router = Router(_config(), _backlog())
client = _mock_client()
await router.detach_all(client) # should not raise
@pytest.mark.asyncio
async def test_multiple_clients(self) -> None:
router = Router(_config(), _backlog())
c1 = _mock_client("user1")
c2 = _mock_client("user2")
await router.attach_all(c1)
await router.attach_all(c2)
assert len(router.clients) == 2
await router.detach_all(c1)
assert len(router.clients) == 1
assert c2 in router.clients
# -- stop_networks -----------------------------------------------------------
class TestStopNetworks:
@pytest.mark.asyncio
async def test_stops_all(self) -> None:
router = Router(_config(), _backlog())
n1 = _mock_network("libera")
n2 = _mock_network("oftc")
router.networks = {"libera": n1, "oftc": n2}
await router.stop_networks()
n1.stop.assert_awaited_once()
n2.stop.assert_awaited_once()
# -- route_client_message (outbound: client -> network) ----------------------
class TestRouteClientMessage:
@pytest.mark.asyncio
async def test_privmsg_to_channel(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("PRIVMSG", ["#test/libera", "hello"])
await router.route_client_message(msg)
net.send.assert_awaited_once()
sent = net.send.call_args[0][0]
assert sent.command == "PRIVMSG"
assert sent.params == ["#test", "hello"]
@pytest.mark.asyncio
async def test_privmsg_to_nick(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("PRIVMSG", ["user123/libera", "hi there"])
await router.route_client_message(msg)
sent = net.send.call_args[0][0]
assert sent.params == ["user123", "hi there"]
@pytest.mark.asyncio
async def test_no_namespace_dropped(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("PRIVMSG", ["#test", "hello"])
await router.route_client_message(msg)
net.send.assert_not_awaited()
@pytest.mark.asyncio
async def test_unknown_network_dropped(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("PRIVMSG", ["#test/fakenet", "hello"])
await router.route_client_message(msg)
net.send.assert_not_awaited()
@pytest.mark.asyncio
async def test_disconnected_network_dropped(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera", connected=False)
router.networks["libera"] = net
msg = _msg("PRIVMSG", ["#test/libera", "hello"])
await router.route_client_message(msg)
net.send.assert_not_awaited()
@pytest.mark.asyncio
async def test_empty_params_ignored(self) -> None:
router = Router(_config(), _backlog())
msg = _msg("PRIVMSG")
await router.route_client_message(msg) # should not raise
@pytest.mark.asyncio
async def test_join_single_channel(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("JOIN", ["#dev/libera"])
await router.route_client_message(msg)
sent = net.send.call_args[0][0]
assert sent.command == "JOIN"
assert sent.params == ["#dev"]
@pytest.mark.asyncio
async def test_join_comma_separated_same_network(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("JOIN", ["#a/libera,#b/libera"])
await router.route_client_message(msg)
sent = net.send.call_args[0][0]
assert sent.params[0] == "#a,#b"
@pytest.mark.asyncio
async def test_join_comma_separated_multi_network(self) -> None:
router = Router(_config(), _backlog())
n1 = _mock_network("libera")
n2 = _mock_network("oftc")
router.networks = {"libera": n1, "oftc": n2}
msg = _msg("JOIN", ["#a/libera,#b/oftc"])
await router.route_client_message(msg)
assert n1.send.await_count == 1
assert n2.send.await_count == 1
s1 = n1.send.call_args[0][0]
s2 = n2.send.call_args[0][0]
assert s1.params[0] == "#a"
assert s2.params[0] == "#b"
@pytest.mark.asyncio
async def test_part(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("PART", ["#dev/libera", "leaving"])
await router.route_client_message(msg)
sent = net.send.call_args[0][0]
assert sent.command == "PART"
assert sent.params == ["#dev", "leaving"]
@pytest.mark.asyncio
async def test_kick(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("KICK", ["#test/libera", "baduser/libera", "bye"])
await router.route_client_message(msg)
sent = net.send.call_args[0][0]
assert sent.command == "KICK"
assert sent.params == ["#test", "baduser", "bye"]
@pytest.mark.asyncio
async def test_kick_no_namespace_dropped(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("KICK", ["#test", "baduser"])
await router.route_client_message(msg)
net.send.assert_not_awaited()
@pytest.mark.asyncio
async def test_invite(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("INVITE", ["user/libera", "#test/libera"])
await router.route_client_message(msg)
sent = net.send.call_args[0][0]
assert sent.command == "INVITE"
assert sent.params == ["user", "#test"]
@pytest.mark.asyncio
async def test_invite_network_from_either_param(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
# Network suffix only on the nick, not the channel
msg = _msg("INVITE", ["user/libera", "#test"])
await router.route_client_message(msg)
net.send.assert_awaited_once()
@pytest.mark.asyncio
async def test_mode_channel(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("MODE", ["#test/libera", "+o", "nick"])
await router.route_client_message(msg)
sent = net.send.call_args[0][0]
assert sent.params == ["#test", "+o", "nick"]
@pytest.mark.asyncio
async def test_who(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("WHO", ["#test/libera"])
await router.route_client_message(msg)
sent = net.send.call_args[0][0]
assert sent.params == ["#test"]
@pytest.mark.asyncio
async def test_notice(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("NOTICE", ["user/libera", "you've been warned"])
await router.route_client_message(msg)
sent = net.send.call_args[0][0]
assert sent.params == ["user", "you've been warned"]
@pytest.mark.asyncio
async def test_topic(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("TOPIC", ["#test/libera", "new topic"])
await router.route_client_message(msg)
sent = net.send.call_args[0][0]
assert sent.params == ["#test", "new topic"]
@pytest.mark.asyncio
async def test_preserves_tags(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("PRIVMSG", ["#test/libera", "hi"],
tags={"label": "abc"})
await router.route_client_message(msg)
sent = net.send.call_args[0][0]
assert sent.tags == {"label": "abc"}
@pytest.mark.asyncio
async def test_preserves_prefix(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera")
router.networks["libera"] = net
msg = _msg("PRIVMSG", ["#test/libera", "hi"], prefix="me!u@h")
await router.route_client_message(msg)
sent = net.send.call_args[0][0]
assert sent.prefix == "me!u@h"
# -- _dispatch (inbound: network -> clients) ---------------------------------
class TestDispatch:
@pytest.mark.asyncio
async def test_delivers_to_all_clients(self) -> None:
router = Router(_config(), _backlog())
net = _mock_network("libera", nick="bot")
router.networks["libera"] = net
c1 = _mock_client("user1")
c2 = _mock_client("user2")
router.clients = [c1, c2]
msg = _msg("PRIVMSG", ["#test", "hello"], prefix="sender!u@h")
await router._dispatch("libera", msg)
assert c1.write.call_count == 1
assert c2.write.call_count == 1
@pytest.mark.asyncio
async def test_suppressed_not_delivered(self) -> None:
router = Router(_config(), _backlog())
router.networks["libera"] = _mock_network("libera")
client = _mock_client()
router.clients = [client]
msg = _msg("001", ["nick", "Welcome"])
await router._dispatch("libera", msg)
client.write.assert_not_called()
@pytest.mark.asyncio
async def test_namespaces_channel(self) -> None:
router = Router(_config(), _backlog())
router.networks["libera"] = _mock_network("libera", nick="bot")
client = _mock_client("me")
router.clients = [client]
msg = _msg("PRIVMSG", ["#test", "hello"], prefix="user!i@h")
await router._dispatch("libera", msg)
written = client.write.call_args[0][0]
assert b"#test/libera" in written
@pytest.mark.asyncio
async def test_namespaces_prefix(self) -> None:
router = Router(_config(), _backlog())
router.networks["libera"] = _mock_network("libera", nick="bot")
client = _mock_client("me")
router.clients = [client]
msg = _msg("PRIVMSG", ["#test", "hello"], prefix="sender!i@h")
await router._dispatch("libera", msg)
written = client.write.call_args[0][0]
assert b"sender/libera" in written
@pytest.mark.asyncio
async def test_own_nick_rewritten_to_client_nick(self) -> None:
router = Router(_config(), _backlog())
router.networks["libera"] = _mock_network("libera", nick="bot")
client = _mock_client("clientnick")
router.clients = [client]
msg = _msg("PRIVMSG", ["#test", "hello"], prefix="bot!i@h")
await router._dispatch("libera", msg)
written = client.write.call_args[0][0]
assert b"clientnick" in written
assert b"bot/libera" not in written
@pytest.mark.asyncio
async def test_client_write_error_handled(self) -> None:
router = Router(_config(), _backlog())
router.networks["libera"] = _mock_network("libera", nick="bot")
bad_client = _mock_client()
bad_client.write.side_effect = ConnectionResetError
good_client = _mock_client()
router.clients = [bad_client, good_client]
msg = _msg("PRIVMSG", ["#test", "hello"], prefix="user!i@h")
await router._dispatch("libera", msg)
# Bad client raised, but good client still received
good_client.write.assert_called_once()
@pytest.mark.asyncio
async def test_no_clients_no_error(self) -> None:
router = Router(_config(), _backlog())
router.networks["libera"] = _mock_network("libera")
router.clients = []
msg = _msg("PRIVMSG", ["#test", "hello"], prefix="user!i@h")
await router._dispatch("libera", msg) # should not raise
# -- _on_network_status ------------------------------------------------------
class TestOnNetworkStatus:
def test_broadcasts_to_all_clients(self) -> None:
router = Router(_config(), _backlog())
c1 = _mock_client()
c2 = _mock_client()
router.clients = [c1, c2]
router._on_network_status("libera", "connection stable")
assert c1.write.call_count == 1
assert c2.write.call_count == 1
written = c1.write.call_args[0][0]
assert b"[libera] connection stable" in written
assert b"bouncer" in written # prefix
def test_client_error_does_not_propagate(self) -> None:
router = Router(_config(), _backlog())
bad = _mock_client()
bad.write.side_effect = ConnectionResetError
good = _mock_client()
router.clients = [bad, good]
router._on_network_status("libera", "test")
good.write.assert_called_once()
# -- _on_network_message (sync -> async bridge) -----------------------------
class TestOnNetworkMessage:
@pytest.mark.asyncio
async def test_creates_dispatch_task(self) -> None:
router = Router(_config(), _backlog())
router.networks["libera"] = _mock_network("libera")
with patch.object(router, "_dispatch", new_callable=AsyncMock) as mock_disp:
msg = _msg("PRIVMSG", ["#test", "hi"], prefix="u!i@h")
router._on_network_message("libera", msg)
# Let the task run
await asyncio.sleep(0)
mock_disp.assert_awaited_once_with("libera", msg)
# -- _replay_backlog ---------------------------------------------------------
class TestReplayBacklog:
@pytest.mark.asyncio
async def test_empty_backlog(self) -> None:
bl = _backlog()
router = Router(_config(), bl)
router.networks["libera"] = _mock_network("libera", nick="bot")
client = _mock_client()
await router._replay_backlog(client, "libera")
client.write.assert_not_called()
@pytest.mark.asyncio
async def test_replays_messages(self) -> None:
bl = _backlog()
entry1 = MagicMock()
entry1.id = 1
entry1.command = "PRIVMSG"
entry1.target = "#test"
entry1.content = "hello"
entry1.sender = "user!i@h"
entry2 = MagicMock()
entry2.id = 2
entry2.command = "PRIVMSG"
entry2.target = "#test"
entry2.content = "world"
entry2.sender = "other!i@h"
bl.replay.return_value = [entry1, entry2]
router = Router(_config(), bl)
router.networks["libera"] = _mock_network("libera", nick="bot")
client = _mock_client("me")
await router._replay_backlog(client, "libera")
assert client.write.call_count == 2
# Verify namespaced
first = client.write.call_args_list[0][0][0]
assert b"#test/libera" in first
# Should mark last seen
bl.mark_seen.assert_awaited_once_with("libera", 2)
@pytest.mark.asyncio
async def test_replays_since_last_seen(self) -> None:
bl = _backlog()
bl.get_last_seen.return_value = 42
router = Router(_config(), bl)
router.networks["libera"] = _mock_network("libera", nick="bot")
client = _mock_client()
await router._replay_backlog(client, "libera")
bl.replay.assert_awaited_once_with("libera", since_id=42)
@pytest.mark.asyncio
async def test_suppressed_skipped_during_replay(self) -> None:
bl = _backlog()
entry = MagicMock()
entry.id = 1
entry.command = "NOTICE"
entry.target = "nick"
entry.content = "\x01VERSION mIRC\x01"
entry.sender = "server.example.com" # no '!' -> server notice
bl.replay.return_value = [entry]
router = Router(_config(), bl)
router.networks["libera"] = _mock_network("libera", nick="bot")
client = _mock_client()
await router._replay_backlog(client, "libera")
client.write.assert_not_called()
# Still marks seen even if all suppressed
bl.mark_seen.assert_awaited_once_with("libera", 1)
@pytest.mark.asyncio
async def test_client_error_stops_replay(self) -> None:
bl = _backlog()
entry1 = MagicMock()
entry1.id = 1
entry1.command = "PRIVMSG"
entry1.target = "#test"
entry1.content = "msg1"
entry1.sender = "user!i@h"
entry2 = MagicMock()
entry2.id = 2
entry2.command = "PRIVMSG"
entry2.target = "#test"
entry2.content = "msg2"
entry2.sender = "user!i@h"
bl.replay.return_value = [entry1, entry2]
router = Router(_config(), bl)
router.networks["libera"] = _mock_network("libera", nick="bot")
client = _mock_client()
client.write.side_effect = [ConnectionResetError, None]
await router._replay_backlog(client, "libera")
# Should have stopped after first error
assert client.write.call_count == 1
# -- BACKLOG_COMMANDS constant -----------------------------------------------
class TestBacklogCommands:
def test_expected_commands(self) -> None:
assert "PRIVMSG" in BACKLOG_COMMANDS
assert "NOTICE" in BACKLOG_COMMANDS
assert "TOPIC" in BACKLOG_COMMANDS
assert "KICK" in BACKLOG_COMMANDS
assert "MODE" in BACKLOG_COMMANDS
def test_join_not_in_backlog(self) -> None:
assert "JOIN" not in BACKLOG_COMMANDS
assert "PART" not in BACKLOG_COMMANDS
assert "QUIT" not in BACKLOG_COMMANDS
# -- server-time tag injection -----------------------------------------------
class TestServerTimeDispatch:
@pytest.mark.asyncio
async def test_injects_time_tag_when_missing(self) -> None:
router = Router(_config(), _backlog())
router.networks["libera"] = _mock_network("libera", nick="bot")
client = _mock_client("me")
router.clients = [client]
msg = _msg("PRIVMSG", ["#test", "hello"], prefix="user!i@h")
await router._dispatch("libera", msg)
assert "time" in msg.tags
# Verify ISO8601 format
assert msg.tags["time"].endswith("Z")
assert "T" in msg.tags["time"]
@pytest.mark.asyncio
async def test_preserves_existing_time_tag(self) -> None:
router = Router(_config(), _backlog())
router.networks["libera"] = _mock_network("libera", nick="bot")
client = _mock_client("me")
router.clients = [client]
original_time = "2025-01-15T12:00:00.000000Z"
msg = _msg("PRIVMSG", ["#test", "hello"], prefix="user!i@h",
tags={"time": original_time})
await router._dispatch("libera", msg)
assert msg.tags["time"] == original_time
class TestServerTimeReplay:
@pytest.mark.asyncio
async def test_replay_injects_timestamp(self) -> None:
bl = _backlog()
entry = MagicMock()
entry.id = 1
entry.command = "PRIVMSG"
entry.target = "#test"
entry.content = "hello"
entry.sender = "user!i@h"
entry.timestamp = 1705320000.0 # 2024-01-15T12:00:00Z
bl.replay.return_value = [entry]
router = Router(_config(), bl)
router.networks["libera"] = _mock_network("libera", nick="bot")
client = _mock_client("me")
await router._replay_backlog(client, "libera")
written = client.write.call_args[0][0]
# The time tag should be in the wire format
assert b"time=" in written
# -- push notifications ------------------------------------------------------
class TestNotifications:
@pytest.mark.asyncio
async def test_notification_triggered_on_pm_no_clients(self) -> None:
cfg = _config()
cfg.bouncer.notify_url = "https://ntfy.sh/test"
router = Router(cfg, _backlog())
net = _mock_network("libera", nick="bot")
router.networks["libera"] = net
router.clients = [] # no clients
with patch.object(router._notifier, "should_notify", return_value=True):
with patch.object(router._notifier, "send", new_callable=AsyncMock) as mock_send:
msg = _msg("PRIVMSG", ["bot", "hello bot"], prefix="user!i@h")
await router._dispatch("libera", msg)
# Let fire-and-forget task run
await asyncio.sleep(0)
mock_send.assert_awaited_once_with("libera", "user", "bot", "hello bot")
@pytest.mark.asyncio
async def test_notification_not_triggered_with_clients(self) -> None:
cfg = _config()
cfg.bouncer.notify_url = "https://ntfy.sh/test"
router = Router(cfg, _backlog())
net = _mock_network("libera", nick="bot")
router.networks["libera"] = net
client = _mock_client()
router.clients = [client]
with patch.object(router._notifier, "send", new_callable=AsyncMock) as mock_send:
msg = _msg("PRIVMSG", ["bot", "hello bot"], prefix="user!i@h")
await router._dispatch("libera", msg)
await asyncio.sleep(0)
mock_send.assert_not_awaited()
@pytest.mark.asyncio
async def test_notification_respects_should_notify(self) -> None:
cfg = _config()
cfg.bouncer.notify_url = "https://ntfy.sh/test"
router = Router(cfg, _backlog())
net = _mock_network("libera", nick="bot")
router.networks["libera"] = net
router.clients = []
with patch.object(router._notifier, "should_notify", return_value=False):
with patch.object(router._notifier, "send", new_callable=AsyncMock) as mock_send:
msg = _msg("PRIVMSG", ["#channel", "random msg"], prefix="user!i@h")
await router._dispatch("libera", msg)
await asyncio.sleep(0)
mock_send.assert_not_awaited()
@pytest.mark.asyncio
async def test_notification_disabled_skips(self) -> None:
cfg = _config()
cfg.bouncer.notify_url = "" # disabled
router = Router(cfg, _backlog())
net = _mock_network("libera", nick="bot")
router.networks["libera"] = net
router.clients = []
with patch.object(router._notifier, "send", new_callable=AsyncMock) as mock_send:
msg = _msg("PRIVMSG", ["bot", "hello"], prefix="user!i@h")
await router._dispatch("libera", msg)
await asyncio.sleep(0)
mock_send.assert_not_awaited()
Reference in New Issue View Git Blame Copy Permalink