Network control (CONNECT, DISCONNECT, RECONNECT, NICK, RAW), visibility (CHANNELS, CLIENTS, BACKLOG, VERSION), config management (REHASH, ADDNETWORK, DELNETWORK, AUTOJOIN), and NickServ operations (IDENTIFY, REGISTER, DROPCREDS). Total command count: 22. Adds stats()/db_size() to Backlog, add_network()/remove_network() to Router, and _connected_at timestamp to Client. 74 command tests. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
785 lines
29 KiB
Python
785 lines
29 KiB
Python
"""Tests for bouncer control commands."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from bouncer import commands
|
|
from bouncer.network import State
|
|
|
|
|
|
def _make_network(name: str, state: State, nick: str = "testnick",
|
|
host: str | None = None, channels: set[str] | None = None,
|
|
topics: dict[str, str] | None = None) -> MagicMock:
|
|
"""Create a mock Network."""
|
|
net = MagicMock()
|
|
net.cfg.name = name
|
|
net.cfg.host = f"irc.{name}.chat"
|
|
net.cfg.port = 6697
|
|
net.cfg.tls = True
|
|
net.cfg.channels = list(channels) if channels else []
|
|
net.cfg.nick = nick
|
|
net.cfg.password = None
|
|
net.state = state
|
|
net.nick = nick
|
|
net.visible_host = host
|
|
net.channels = channels or set()
|
|
net.topics = topics or {}
|
|
net.names = {}
|
|
net._reconnect_attempt = 0
|
|
net.connected = state not in (State.DISCONNECTED, State.CONNECTING)
|
|
net.ready = state == State.READY
|
|
net.start = AsyncMock()
|
|
net.stop = AsyncMock()
|
|
net.send = AsyncMock()
|
|
net.send_raw = AsyncMock()
|
|
net._nickserv_register = AsyncMock()
|
|
return net
|
|
|
|
|
|
def _make_router(*networks: MagicMock) -> MagicMock:
|
|
"""Create a mock Router with the given networks."""
|
|
router = MagicMock()
|
|
router.networks = {n.cfg.name: n for n in networks}
|
|
router.network_names.return_value = [n.cfg.name for n in networks]
|
|
router.get_network = lambda name: router.networks.get(name)
|
|
router.backlog = AsyncMock()
|
|
router.add_network = AsyncMock()
|
|
router.remove_network = AsyncMock(return_value=True)
|
|
router.config = MagicMock()
|
|
return router
|
|
|
|
|
|
def _make_client(nick: str = "testuser") -> MagicMock:
|
|
"""Create a mock Client."""
|
|
client = MagicMock()
|
|
client.nick = nick
|
|
client._connected_at = time.time() - 120
|
|
client._addr = ("127.0.0.1", 54321)
|
|
return client
|
|
|
|
|
|
class TestHelp:
|
|
@pytest.mark.asyncio
|
|
async def test_help_lists_commands(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("HELP", router, client)
|
|
assert lines[0] == "[HELP]"
|
|
assert any("STATUS" in line for line in lines)
|
|
assert any("UPTIME" in line for line in lines)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_input_shows_help(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("", router, client)
|
|
assert lines[0] == "[HELP]"
|
|
|
|
|
|
class TestStatus:
|
|
@pytest.mark.asyncio
|
|
async def test_status_no_networks(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("STATUS", router, client)
|
|
assert lines[0] == "[STATUS]"
|
|
assert "(no networks configured)" in lines[1]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_ready_network(self) -> None:
|
|
net = _make_network("libera", State.READY, nick="fabesune", host="user/fabesune")
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("STATUS", router, client)
|
|
assert lines[0] == "[STATUS]"
|
|
assert "ready" in lines[1]
|
|
assert "fabesune" in lines[1]
|
|
assert "user/fabesune" in lines[1]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_connecting_shows_attempt(self) -> None:
|
|
net = _make_network("hackint", State.CONNECTING)
|
|
net._reconnect_attempt = 3
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("STATUS", router, client)
|
|
assert "connecting" in lines[1]
|
|
assert "attempt 3" in lines[1]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_case_insensitive(self) -> None:
|
|
net = _make_network("libera", State.READY, nick="testnick")
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("status", router, client)
|
|
assert lines[0] == "[STATUS]"
|
|
|
|
|
|
class TestInfo:
|
|
@pytest.mark.asyncio
|
|
async def test_info_missing_arg(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("INFO", router, client)
|
|
assert "Usage" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_info_unknown_network(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("INFO fakenet", router, client)
|
|
assert "Unknown network" in lines[0]
|
|
assert "libera" in lines[1]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_info_valid_network(self) -> None:
|
|
net = _make_network("libera", State.READY, nick="fabesune",
|
|
host="user/fabesune", channels={"#test", "#dev"})
|
|
router = _make_router(net)
|
|
router.backlog.list_nickserv_creds.return_value = [
|
|
("libera", "fabesune", "test@mail.tm", "user/fabesune", 1700000000.0, "verified"),
|
|
]
|
|
client = _make_client()
|
|
lines = await commands.dispatch("INFO libera", router, client)
|
|
assert lines[0] == "[INFO] libera"
|
|
assert any("ready" in line for line in lines)
|
|
assert any("fabesune" in line for line in lines)
|
|
assert any("#dev" in line or "#test" in line for line in lines)
|
|
assert any("verified" in line for line in lines)
|
|
|
|
|
|
class TestUptime:
|
|
@pytest.mark.asyncio
|
|
async def test_uptime(self) -> None:
|
|
commands.STARTUP_TIME = time.time() - 3661 # 1h 1m 1s
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("UPTIME", router, client)
|
|
assert lines[0].startswith("[UPTIME]")
|
|
assert "1h" in lines[0]
|
|
assert "1m" in lines[0]
|
|
assert "1s" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uptime_unknown(self) -> None:
|
|
commands.STARTUP_TIME = 0.0
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("UPTIME", router, client)
|
|
assert "unknown" in lines[0]
|
|
|
|
|
|
class TestNetworks:
|
|
@pytest.mark.asyncio
|
|
async def test_networks_empty(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("NETWORKS", router, client)
|
|
assert lines[0] == "[NETWORKS]"
|
|
assert "(none)" in lines[1]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_networks_lists_all(self) -> None:
|
|
libera = _make_network("libera", State.READY)
|
|
oftc = _make_network("oftc", State.CONNECTING)
|
|
router = _make_router(libera, oftc)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("NETWORKS", router, client)
|
|
assert lines[0] == "[NETWORKS]"
|
|
assert any("libera" in line and "ready" in line for line in lines[1:])
|
|
assert any("oftc" in line and "connecting" in line for line in lines[1:])
|
|
|
|
|
|
class TestCreds:
|
|
@pytest.mark.asyncio
|
|
async def test_creds_no_backlog(self) -> None:
|
|
router = _make_router()
|
|
router.backlog = None
|
|
client = _make_client()
|
|
lines = await commands.dispatch("CREDS", router, client)
|
|
assert "not available" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_creds_empty(self) -> None:
|
|
router = _make_router()
|
|
router.backlog.list_nickserv_creds.return_value = []
|
|
client = _make_client()
|
|
lines = await commands.dispatch("CREDS", router, client)
|
|
assert "no stored credentials" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_creds_lists_entries(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
router.backlog.list_nickserv_creds.return_value = [
|
|
("libera", "fabesune", "test@mail.tm", "user/fabesune", 1700000000.0, "verified"),
|
|
("libera", "oldnick", "old@mail.tm", "old/host", 1699000000.0, "pending"),
|
|
]
|
|
client = _make_client()
|
|
lines = await commands.dispatch("CREDS libera", router, client)
|
|
assert lines[0] == "[CREDS]"
|
|
assert any("+" in line and "fabesune" in line and "verified" in line for line in lines)
|
|
assert any("~" in line and "oldnick" in line and "pending" in line for line in lines)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_creds_unknown_network(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("CREDS fakenet", router, client)
|
|
assert "Unknown network" in lines[0]
|
|
|
|
|
|
class TestConnect:
|
|
@pytest.mark.asyncio
|
|
async def test_connect_missing_arg(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("CONNECT", router, client)
|
|
assert "Usage" in lines[0] or "provide" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_unknown_network(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("CONNECT fakenet", router, client)
|
|
assert "Unknown network" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_already_connected(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("CONNECT libera", router, client)
|
|
assert "already" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_disconnected(self) -> None:
|
|
net = _make_network("libera", State.DISCONNECTED)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("CONNECT libera", router, client)
|
|
assert "[CONNECT]" in lines[0]
|
|
assert "starting" in lines[0]
|
|
|
|
|
|
class TestDisconnect:
|
|
@pytest.mark.asyncio
|
|
async def test_disconnect_ready(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("DISCONNECT libera", router, client)
|
|
assert "[DISCONNECT]" in lines[0]
|
|
net.stop.assert_awaited_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disconnect_already_disconnected(self) -> None:
|
|
net = _make_network("libera", State.DISCONNECTED)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("DISCONNECT libera", router, client)
|
|
assert "already disconnected" in lines[0]
|
|
|
|
|
|
class TestReconnect:
|
|
@pytest.mark.asyncio
|
|
async def test_reconnect(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("RECONNECT libera", router, client)
|
|
assert "[RECONNECT]" in lines[0]
|
|
net.stop.assert_awaited_once()
|
|
|
|
|
|
class TestNick:
|
|
@pytest.mark.asyncio
|
|
async def test_nick_missing_args(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("NICK", router, client)
|
|
assert "Usage" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nick_missing_nick(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("NICK libera", router, client)
|
|
assert "Usage" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nick_change(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("NICK libera newnick", router, client)
|
|
assert "[NICK]" in lines[0]
|
|
net.send_raw.assert_awaited_once_with("NICK", "newnick")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nick_not_connected(self) -> None:
|
|
net = _make_network("libera", State.DISCONNECTED)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("NICK libera newnick", router, client)
|
|
assert "not connected" in lines[0]
|
|
|
|
|
|
class TestRaw:
|
|
@pytest.mark.asyncio
|
|
async def test_raw_missing_args(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("RAW", router, client)
|
|
assert "Usage" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_raw_send(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("RAW libera WHOIS testuser", router, client)
|
|
assert "[RAW]" in lines[0]
|
|
net.send.assert_awaited_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_raw_not_connected(self) -> None:
|
|
net = _make_network("libera", State.DISCONNECTED)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("RAW libera WHOIS testuser", router, client)
|
|
assert "not connected" in lines[0]
|
|
|
|
|
|
class TestChannels:
|
|
@pytest.mark.asyncio
|
|
async def test_channels_all(self) -> None:
|
|
net = _make_network("libera", State.READY, channels={"#test", "#dev"})
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("CHANNELS", router, client)
|
|
assert lines[0] == "[CHANNELS]"
|
|
assert any("#test" in line for line in lines)
|
|
assert any("#dev" in line for line in lines)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_channels_specific_network(self) -> None:
|
|
net = _make_network("libera", State.READY, channels={"#test"})
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("CHANNELS libera", router, client)
|
|
assert lines[0] == "[CHANNELS]"
|
|
assert any("#test" in line for line in lines)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_channels_unknown_network(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("CHANNELS fakenet", router, client)
|
|
assert "Unknown network" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_channels_with_topics(self) -> None:
|
|
net = _make_network("libera", State.READY, channels={"#test"},
|
|
topics={"#test": "Welcome to test"})
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("CHANNELS libera", router, client)
|
|
assert any("Welcome to test" in line for line in lines)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_channels_empty(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("CHANNELS libera", router, client)
|
|
assert any("(none)" in line for line in lines)
|
|
|
|
|
|
class TestClients:
|
|
@pytest.mark.asyncio
|
|
async def test_clients_empty(self) -> None:
|
|
router = _make_router()
|
|
router.clients = []
|
|
client = _make_client()
|
|
lines = await commands.dispatch("CLIENTS", router, client)
|
|
assert lines[0] == "[CLIENTS]"
|
|
assert "(none)" in lines[1]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clients_lists_connected(self) -> None:
|
|
router = _make_router()
|
|
c1 = _make_client("user1")
|
|
c2 = _make_client("user2")
|
|
router.clients = [c1, c2]
|
|
client = _make_client()
|
|
lines = await commands.dispatch("CLIENTS", router, client)
|
|
assert lines[0] == "[CLIENTS]"
|
|
assert any("user1" in line for line in lines)
|
|
assert any("user2" in line for line in lines)
|
|
assert any("connected" in line for line in lines)
|
|
|
|
|
|
class TestBacklog:
|
|
@pytest.mark.asyncio
|
|
async def test_backlog_no_backlog(self) -> None:
|
|
router = _make_router()
|
|
router.backlog = None
|
|
client = _make_client()
|
|
lines = await commands.dispatch("BACKLOG", router, client)
|
|
assert "not available" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backlog_stats(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
router.backlog.stats.return_value = [("libera", 1500)]
|
|
router.backlog.db_size.return_value = 2_097_152 # 2 MB
|
|
client = _make_client()
|
|
lines = await commands.dispatch("BACKLOG", router, client)
|
|
assert lines[0] == "[BACKLOG]"
|
|
assert any("1,500" in line for line in lines)
|
|
assert any("2.0 MB" in line for line in lines)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backlog_specific_network(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
router.backlog.stats.return_value = [("libera", 42)]
|
|
router.backlog.db_size.return_value = 4096
|
|
client = _make_client()
|
|
lines = await commands.dispatch("BACKLOG libera", router, client)
|
|
assert lines[0] == "[BACKLOG]"
|
|
assert any("42" in line for line in lines)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backlog_unknown_network(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("BACKLOG fakenet", router, client)
|
|
assert "Unknown network" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backlog_empty(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
router.backlog.stats.return_value = []
|
|
router.backlog.db_size.return_value = 1024
|
|
client = _make_client()
|
|
lines = await commands.dispatch("BACKLOG", router, client)
|
|
assert any("no messages" in line for line in lines)
|
|
|
|
|
|
class TestVersion:
|
|
@pytest.mark.asyncio
|
|
async def test_version(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("VERSION", router, client)
|
|
assert "[VERSION]" in lines[0]
|
|
assert "bouncer" in lines[0]
|
|
assert "Python" in lines[0]
|
|
|
|
|
|
class TestRehash:
|
|
@pytest.mark.asyncio
|
|
async def test_rehash_no_config_path(self) -> None:
|
|
commands.CONFIG_PATH = None
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("REHASH", router, client)
|
|
assert "config path not set" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rehash_config_error(self) -> None:
|
|
commands.CONFIG_PATH = Path("/nonexistent/config.toml")
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("REHASH", router, client)
|
|
assert "config error" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rehash_adds_and_removes(self) -> None:
|
|
from bouncer.config import BouncerConfig, Config, NetworkConfig, ProxyConfig
|
|
|
|
old_net = _make_network("libera", State.READY)
|
|
router = _make_router(old_net)
|
|
|
|
new_cfg = Config(
|
|
bouncer=BouncerConfig(),
|
|
proxy=ProxyConfig(),
|
|
networks={
|
|
"oftc": NetworkConfig(name="oftc", host="irc.oftc.net", port=6697, tls=True),
|
|
},
|
|
)
|
|
|
|
commands.CONFIG_PATH = Path("/tmp/test.toml")
|
|
with patch("bouncer.config.load", return_value=new_cfg):
|
|
client = _make_client()
|
|
lines = await commands.dispatch("REHASH", router, client)
|
|
|
|
assert lines[0] == "[REHASH]"
|
|
assert any("removed: libera" in line for line in lines)
|
|
assert any("added: oftc" in line for line in lines)
|
|
router.remove_network.assert_awaited()
|
|
router.add_network.assert_awaited()
|
|
|
|
|
|
class TestAddNetwork:
|
|
@pytest.mark.asyncio
|
|
async def test_addnetwork_missing_args(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("ADDNETWORK", router, client)
|
|
assert "Usage" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_addnetwork_missing_host(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("ADDNETWORK testnet port=6667", router, client)
|
|
assert "Required" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_addnetwork_already_exists(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("ADDNETWORK libera host=irc.libera.chat", router, client)
|
|
assert "already exists" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_addnetwork_success(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch(
|
|
"ADDNETWORK testnet host=irc.test.com port=6697 tls=yes nick=mynick channels=#a,#b",
|
|
router, client,
|
|
)
|
|
assert "[ADDNETWORK]" in lines[0]
|
|
assert "testnet" in lines[0]
|
|
router.add_network.assert_awaited_once()
|
|
cfg = router.add_network.call_args[0][0]
|
|
assert cfg.name == "testnet"
|
|
assert cfg.host == "irc.test.com"
|
|
assert cfg.port == 6697
|
|
assert cfg.tls is True
|
|
assert cfg.nick == "mynick"
|
|
assert cfg.channels == ["#a", "#b"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_addnetwork_slash_in_name(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("ADDNETWORK test/net host=h", router, client)
|
|
assert "must not contain" in lines[0]
|
|
|
|
|
|
class TestDelNetwork:
|
|
@pytest.mark.asyncio
|
|
async def test_delnetwork_missing_arg(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("DELNETWORK", router, client)
|
|
assert "Usage" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delnetwork_unknown(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("DELNETWORK fakenet", router, client)
|
|
assert "Unknown network" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delnetwork_success(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("DELNETWORK libera", router, client)
|
|
assert "[DELNETWORK]" in lines[0]
|
|
router.remove_network.assert_awaited_once_with("libera")
|
|
|
|
|
|
class TestAutojoin:
|
|
@pytest.mark.asyncio
|
|
async def test_autojoin_missing_args(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("AUTOJOIN", router, client)
|
|
assert "Usage" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_autojoin_add(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
net.cfg.channels = ["#test"]
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("AUTOJOIN libera +#dev", router, client)
|
|
assert "[AUTOJOIN]" in lines[0]
|
|
assert any("added: #dev" in line for line in lines)
|
|
assert "#dev" in net.cfg.channels
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_autojoin_remove(self) -> None:
|
|
net = _make_network("libera", State.READY, channels={"#test"})
|
|
net.cfg.channels = ["#test"]
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("AUTOJOIN libera -#test", router, client)
|
|
assert any("removed: #test" in line for line in lines)
|
|
assert "#test" not in net.cfg.channels
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_autojoin_remove_missing(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
net.cfg.channels = []
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("AUTOJOIN libera -#missing", router, client)
|
|
assert any("not in autojoin" in line for line in lines)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_autojoin_invalid_spec(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("AUTOJOIN libera #test", router, client)
|
|
assert "must start with" in lines[0]
|
|
|
|
|
|
class TestIdentify:
|
|
@pytest.mark.asyncio
|
|
async def test_identify_missing_arg(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("IDENTIFY", router, client)
|
|
assert "Usage" in lines[0] or "provide" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_identify_not_connected(self) -> None:
|
|
net = _make_network("libera", State.DISCONNECTED)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("IDENTIFY libera", router, client)
|
|
assert "not connected" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_identify_no_creds(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
router.backlog.get_nickserv_creds_by_network.return_value = None
|
|
client = _make_client()
|
|
lines = await commands.dispatch("IDENTIFY libera", router, client)
|
|
assert "No stored credentials" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_identify_success(self) -> None:
|
|
net = _make_network("libera", State.READY, nick="fabesune")
|
|
router = _make_router(net)
|
|
router.backlog.get_nickserv_creds_by_network.return_value = ("fabesune", "secret123")
|
|
client = _make_client()
|
|
lines = await commands.dispatch("IDENTIFY libera", router, client)
|
|
assert "[IDENTIFY]" in lines[0]
|
|
net.send_raw.assert_awaited_with("PRIVMSG", "NickServ", "IDENTIFY secret123")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_identify_nick_switch(self) -> None:
|
|
net = _make_network("libera", State.READY, nick="randomnick")
|
|
router = _make_router(net)
|
|
router.backlog.get_nickserv_creds_by_network.return_value = ("fabesune", "secret123")
|
|
client = _make_client()
|
|
lines = await commands.dispatch("IDENTIFY libera", router, client)
|
|
assert any("switching nick" in line for line in lines)
|
|
calls = net.send_raw.await_args_list
|
|
assert any(c.args == ("NICK", "fabesune") for c in calls)
|
|
|
|
|
|
class TestRegister:
|
|
@pytest.mark.asyncio
|
|
async def test_register_missing_arg(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("REGISTER", router, client)
|
|
assert "Usage" in lines[0] or "provide" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_not_ready(self) -> None:
|
|
net = _make_network("libera", State.CONNECTING)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("REGISTER libera", router, client)
|
|
assert "not ready" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_success(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("REGISTER libera", router, client)
|
|
assert "[REGISTER]" in lines[0]
|
|
|
|
|
|
class TestDropCreds:
|
|
@pytest.mark.asyncio
|
|
async def test_dropcreds_no_backlog(self) -> None:
|
|
router = _make_router()
|
|
router.backlog = None
|
|
client = _make_client()
|
|
lines = await commands.dispatch("DROPCREDS libera", router, client)
|
|
assert "not available" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dropcreds_missing_arg(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("DROPCREDS", router, client)
|
|
assert "Usage" in lines[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dropcreds_specific_nick(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
client = _make_client()
|
|
lines = await commands.dispatch("DROPCREDS libera fabesune", router, client)
|
|
assert "[DROPCREDS]" in lines[0]
|
|
assert any("deleted: fabesune" in line for line in lines)
|
|
router.backlog.delete_nickserv_creds.assert_awaited_once_with("libera", "fabesune")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dropcreds_all(self) -> None:
|
|
net = _make_network("libera", State.READY)
|
|
router = _make_router(net)
|
|
router.backlog.list_nickserv_creds.return_value = [
|
|
("libera", "nick1", "a@b.c", "", 0.0, "verified"),
|
|
("libera", "nick2", "d@e.f", "", 0.0, "pending"),
|
|
]
|
|
client = _make_client()
|
|
lines = await commands.dispatch("DROPCREDS libera", router, client)
|
|
assert any("deleted: nick1" in line for line in lines)
|
|
assert any("deleted: nick2" in line for line in lines)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dropcreds_unknown_network(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("DROPCREDS fakenet", router, client)
|
|
assert "Unknown network" in lines[0]
|
|
|
|
|
|
class TestUnknownCommand:
|
|
@pytest.mark.asyncio
|
|
async def test_unknown_command(self) -> None:
|
|
router = _make_router()
|
|
client = _make_client()
|
|
lines = await commands.dispatch("FOOBAR", router, client)
|
|
assert "Unknown command" in lines[0]
|
|
assert "HELP" in lines[1]
|