6478c514ad
Users can now inspect bouncer state and manage it from their IRC client by sending PRIVMSG to *bouncer (or bouncer). Supported commands: HELP, STATUS, INFO, UPTIME, NETWORKS, CREDS. Responses arrive as NOTICE messages. All commands are case-insensitive. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
228 lines
8.1 KiB
Python
228 lines
8.1 KiB
Python
"""Tests for bouncer control commands."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
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) -> 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.state = state
|
|
net.nick = nick
|
|
net.visible_host = host
|
|
net.channels = channels or set()
|
|
net._reconnect_attempt = 0
|
|
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()
|
|
return router
|
|
|
|
|
|
def _make_client(nick: str = "testuser") -> MagicMock:
|
|
"""Create a mock Client."""
|
|
client = MagicMock()
|
|
client.nick = nick
|
|
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 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]
|