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

feat: make SOCKS5 proxy configurable per adapter

Browse Source 
Add `proxy` config option to server (IRC), teams, telegram, and mumble
sections. IRC defaults to false (preserving current direct-connect
behavior); all others default to true. The `derp.http` module now
accepts `proxy=True/False` on urlopen, create_connection,
open_connection, and build_opener -- when false, uses stdlib directly.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
user
2026-02-21 21:19:22 +01:00
parent ca46042c41
commit 9d4cb09069
17 changed files with 355 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
TASKS.md
View File

@@ -1,6 +1,19 @@
# derp - Tasks
## Current Sprint -- v2.2.0 Mumble Adapter (2026-02-21)
## Current Sprint -- v2.2.0 Configurable Proxy (2026-02-21)
| Pri | Status | Task |
|-----|--------|------|
| P0 | [x] | `src/derp/http.py` -- `proxy` parameter on all public functions |
| P0 | [x] | `src/derp/config.py` -- `proxy` defaults per adapter section |
| P0 | [x] | `src/derp/irc.py` -- optional SOCKS5 for IRC connections |
| P0 | [x] | `src/derp/telegram.py` -- pass proxy config to HTTP calls |
| P0 | [x] | `src/derp/teams.py` -- pass proxy config to HTTP calls |
| P0 | [x] | `src/derp/mumble.py` -- pass proxy config to TCP calls |
| P1 | [x] | Tests: proxy toggle paths (24 new cases, 1494 total) |
| P2 | [x] | Documentation update (USAGE.md, CHEATSHEET.md, API.md) |
## Previous Sprint -- v2.2.0 Mumble Adapter (2026-02-21)
| Pri | Status | Task |
|-----|--------|------|

11
docs/API.md
View File

@@ -189,14 +189,15 @@ SQLite-backed key-value store. Each plugin gets its own namespace.
## `derp.http` -- HTTP & Network
All outbound traffic routes through the configured SOCKS5 proxy.
HTTP/TCP helpers with optional SOCKS5 proxy routing. All functions accept
a `proxy` parameter (default `True`) to toggle SOCKS5.
| Function | Signature | Description |
|----------|-----------|-------------|
| `urlopen` | `(req, *, timeout=None, context=None, retries=None)` | Proxy-aware HTTP request with connection pooling and retries |
| `build_opener` | `(*handlers, context=None)` | Proxy-aware `urllib.request.build_opener` replacement |
| `create_connection` | `(address, *, timeout=None)` | SOCKS5-proxied `socket.create_connection` with retries |
| `open_connection` | `(host, port, *, timeout=None)` | SOCKS5-proxied `asyncio.open_connection` with retries |
| `urlopen` | `(req, *, timeout=None, context=None, retries=None, proxy=True)` | HTTP request with optional SOCKS5, connection pooling, retries |
| `build_opener` | `(*handlers, context=None, proxy=True)` | Build URL opener, optionally with SOCKS5 handler |
| `create_connection` | `(address, *, timeout=None, proxy=True)` | TCP `socket.create_connection` with optional SOCKS5, retries |
| `open_connection` | `(host, port, *, timeout=None, proxy=True)` | Async `asyncio.open_connection` with optional SOCKS5, retries |
---

9
docs/CHEATSHEET.md
View File

@@ -488,6 +488,7 @@ Auth: HMAC-SHA256 via `X-Signature` header. Starts on IRC connect.
# config/derp.toml
[teams]
enabled = true
proxy = true # SOCKS5 proxy for outbound HTTP
bot_name = "derp"
bind = "127.0.0.1"
port = 8081
@@ -510,6 +511,7 @@ Replies returned as JSON in HTTP response. IRC-only commands (kick, ban, topic)
# config/derp.toml
[telegram]
enabled = true
proxy = true # SOCKS5 proxy for HTTP
bot_token = "123456:ABC-DEF..." # from @BotFather
poll_timeout = 30 # long-poll seconds
admins = [123456789] # Telegram user IDs
@@ -517,8 +519,8 @@ operators = []
trusted = []
```
Long-polling via `getUpdates` -- no public endpoint needed. All HTTP
through SOCKS5 proxy. Strips `@botusername` suffix in groups. Messages
Long-polling via `getUpdates` -- no public endpoint needed. HTTP through
SOCKS5 proxy by default (`proxy = true`). Strips `@botusername` suffix in groups. Messages
split at 4096 chars. IRC-only commands are no-ops. ~90% of plugins work.
## Mumble Integration
@@ -527,6 +529,7 @@ split at 4096 chars. IRC-only commands are no-ops. ~90% of plugins work.
# config/derp.toml
[mumble]
enabled = true
proxy = true # SOCKS5 proxy for TCP
host = "mumble.example.com"
port = 64738
username = "derp"
@@ -537,7 +540,7 @@ operators = []
trusted = []
```
TCP/TLS via SOCKS5 proxy. Text chat only (no voice). Minimal protobuf
TCP/TLS via SOCKS5 proxy by default (`proxy = true`). Text chat only (no voice). Minimal protobuf
codec (no external dep). HTML stripped on receive, escaped on send.
IRC-only commands are no-ops. ~90% of plugins work.

22
docs/USAGE.md
View File

@@ -32,6 +32,7 @@ All settings in `config/derp.toml`.
host = "irc.libera.chat" # IRC server hostname
port = 6697 # Port (6697 = TLS, 6667 = plain)
tls = true # Enable TLS encryption
proxy = false # Route through SOCKS5 proxy (default: false)
nick = "derp" # Bot nickname
user = "derp" # Username (ident)
realname = "derp IRC bot" # Real name field
@@ -1323,6 +1324,7 @@ required -- raw asyncio HTTP, same pattern as the webhook plugin.
```toml
[teams]
enabled = true
proxy = true # Route outbound HTTP through SOCKS5
bot_name = "derp" # outgoing webhook display name
bind = "127.0.0.1" # HTTP listen address
port = 8081 # HTTP listen port
@@ -1418,6 +1420,7 @@ the shared plugin registry. Replies are sent immediately via `sendMessage`.
```toml
[telegram]
enabled = true
proxy = true # Route HTTP through SOCKS5
bot_token = "123456:ABC-DEF..." # from @BotFather
poll_timeout = 30 # long-poll timeout in seconds
admins = [123456789] # Telegram user IDs (numeric)
@@ -1472,19 +1475,20 @@ this automatically: `!help@mybot` becomes `!help`.
### Transport
All HTTP traffic (API calls, long-polling) routes through the SOCKS5
proxy at `127.0.0.1:1080` via `derp.http.urlopen`. No direct outbound
connections are made.
proxy at `127.0.0.1:1080` via `derp.http.urlopen` when `proxy = true`
(default). Set `proxy = false` to connect directly.
## Mumble Integration
Connect derp to a Mumble server via TCP/TLS protobuf control channel.
Text chat only (no voice). All TCP is routed through the SOCKS5 proxy.
No protobuf library dependency -- uses a minimal built-in varint/field
encoder/decoder for the ~7 message types needed.
Text chat only (no voice). TCP is routed through the SOCKS5 proxy when
`proxy = true` (default). No protobuf library dependency -- uses a
minimal built-in varint/field encoder/decoder for the ~7 message types
needed.
### How It Works
The bot connects to the Mumble server over TLS (via SOCKS5), sends
The bot connects to the Mumble server over TLS, sends
Version and Authenticate messages, then enters a read loop. It tracks
channels (ChannelState), users (UserState), and dispatches commands
from TextMessage messages through the shared plugin registry.
@@ -1494,6 +1498,7 @@ from TextMessage messages through the shared plugin registry.
```toml
[mumble]
enabled = true
proxy = true # Route TCP through SOCKS5
host = "mumble.example.com" # Mumble server hostname
port = 64738 # Default Mumble port
username = "derp" # Bot username
@@ -1550,7 +1555,8 @@ unescapes entities. On send, text is HTML-escaped. Action messages use
### Transport
All TCP connections route through the SOCKS5 proxy at `127.0.0.1:1080`
via `derp.http.create_connection`. TLS is applied on top of the proxied
TCP connections route through the SOCKS5 proxy at `127.0.0.1:1080`
via `derp.http.create_connection` when `proxy = true` (default). Set
`proxy = false` to connect directly. TLS is applied on top of the
socket. Mumble commonly uses self-signed certificates, so `tls_verify`
defaults to `false`.

1
src/derp/bot.py
View File

@@ -87,6 +87,7 @@ class Bot:
port=config["server"]["port"],
tls=config["server"]["tls"],
tls_verify=config["server"].get("tls_verify", True),
proxy=config["server"].get("proxy", False),
)
self.nick: str = config["server"]["nick"]
self.prefix: str = config["bot"]["prefix"]

4
src/derp/config.py
View File

@@ -10,6 +10,7 @@ DEFAULTS: dict = {
"host": "irc.libera.chat",
"port": 6697,
"tls": True,
"proxy": False,
"nick": "derp",
"user": "derp",
"realname": "derp IRC bot",
@@ -41,6 +42,7 @@ DEFAULTS: dict = {
},
"teams": {
"enabled": False,
"proxy": True,
"bot_name": "derp",
"bind": "127.0.0.1",
"port": 8081,
@@ -52,6 +54,7 @@ DEFAULTS: dict = {
},
"telegram": {
"enabled": False,
"proxy": True,
"bot_token": "",
"poll_timeout": 30,
"admins": [],
@@ -60,6 +63,7 @@ DEFAULTS: dict = {
},
"mumble": {
"enabled": False,
"proxy": True,
"host": "127.0.0.1",
"port": 64738,
"username": "derp",

77
src/derp/http.py
View File

@@ -1,4 +1,4 @@
"""Proxy-aware HTTP/TCP helpers -- routes outbound traffic through SOCKS5."""
"""HTTP/TCP helpers -- optional SOCKS5 proxy routing for outbound traffic."""
import asyncio
import logging
@@ -85,15 +85,20 @@ class _ProxyHandler(SocksiPyHandler, urllib.request.HTTPSHandler):
# -- Public HTTP interface ---------------------------------------------------
def urlopen(req, *, timeout=None, context=None, retries=None):
"""Proxy-aware drop-in for urllib.request.urlopen.
def urlopen(req, *, timeout=None, context=None, retries=None, proxy=True):
"""HTTP urlopen with optional SOCKS5 proxy.
Uses connection pooling via urllib3 for default requests.
Uses connection pooling via urllib3 for proxied requests.
Falls back to legacy opener for custom SSL context.
When ``proxy=False``, uses stdlib ``urllib.request.urlopen`` directly.
Retries on transient SSL/connection errors with exponential backoff.
"""
max_retries = retries if retries is not None else _MAX_RETRIES
# Direct (no proxy) path
if not proxy:
return _urlopen_direct(req, timeout=timeout, context=context, retries=max_retries)
# Custom SSL context -> fall back to opener (rare: username.py only)
if context is not None:
return _urlopen_legacy(req, timeout=timeout, context=context, retries=max_retries)
@@ -140,6 +145,26 @@ def urlopen(req, *, timeout=None, context=None, retries=None):
time.sleep(delay)
def _urlopen_direct(req, *, timeout=None, context=None, retries=None):
"""Open URL directly without SOCKS5 proxy."""
max_retries = retries if retries is not None else _MAX_RETRIES
kwargs = {}
if timeout is not None:
kwargs["timeout"] = timeout
if context is not None:
kwargs["context"] = context
for attempt in range(max_retries):
try:
return urllib.request.urlopen(req, **kwargs)
except _RETRY_ERRORS as exc:
if attempt + 1 >= max_retries:
raise
delay = 2 ** attempt
_log.debug("urlopen_direct retry %d/%d after %s: %s",
attempt + 1, max_retries, type(exc).__name__, exc)
time.sleep(delay)
def _urlopen_legacy(req, *, timeout=None, context=None, retries=None):
"""Open URL through legacy opener (custom SSL context)."""
max_retries = retries if retries is not None else _MAX_RETRIES
@@ -159,27 +184,32 @@ def _urlopen_legacy(req, *, timeout=None, context=None, retries=None):
time.sleep(delay)
def build_opener(*handlers, context=None):
"""Proxy-aware drop-in for urllib.request.build_opener."""
def build_opener(*handlers, context=None, proxy=True):
"""Build a URL opener, optionally with SOCKS5 proxy."""
if not proxy:
return urllib.request.build_opener(*handlers)
if not handlers and context is None:
return _get_opener()
proxy = _ProxyHandler(context=context)
return urllib.request.build_opener(proxy, *handlers)
proxy_handler = _ProxyHandler(context=context)
return urllib.request.build_opener(proxy_handler, *handlers)
# -- Raw TCP helpers (unchanged) ---------------------------------------------
def create_connection(address, *, timeout=None):
"""SOCKS5-proxied drop-in for socket.create_connection.
def create_connection(address, *, timeout=None, proxy=True):
"""Drop-in for socket.create_connection, optionally through SOCKS5.
Returns a connected socksocket (usable as context manager).
Returns a connected socket (usable as context manager).
Retries on transient connection errors with exponential backoff.
"""
host, port = address
for attempt in range(_MAX_RETRIES):
try:
sock = socks.socksocket()
sock.set_proxy(SOCKS5, _PROXY_ADDR, _PROXY_PORT, rdns=True)
if proxy:
sock = socks.socksocket()
sock.set_proxy(SOCKS5, _PROXY_ADDR, _PROXY_PORT, rdns=True)
else:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if timeout is not None:
sock.settimeout(timeout)
sock.connect((host, port))
@@ -193,12 +223,27 @@ def create_connection(address, *, timeout=None):
time.sleep(delay)
async def open_connection(host, port, *, timeout=None):
"""SOCKS5-proxied drop-in for asyncio.open_connection.
async def open_connection(host, port, *, timeout=None, proxy=True):
"""Async TCP connection, optionally through SOCKS5.
SOCKS5 handshake runs in a thread executor; returns (reader, writer).
When proxied, SOCKS5 handshake runs in a thread executor.
Returns (reader, writer).
Retries on transient connection errors with exponential backoff.
"""
if not proxy:
# Direct asyncio connection
for attempt in range(_MAX_RETRIES):
try:
return await asyncio.open_connection(host, port)
except _RETRY_ERRORS as exc:
if attempt + 1 >= _MAX_RETRIES:
raise
delay = 2 ** attempt
_log.debug("open_connection retry %d/%d after %s: %s",
attempt + 1, _MAX_RETRIES, type(exc).__name__, exc)
await asyncio.sleep(delay)
return # unreachable but satisfies type checker
loop = asyncio.get_running_loop()
def _connect():

27
src/derp/irc.py
View File

@@ -137,11 +137,12 @@ class IRCConnection:
"""Async TCP/TLS connection to an IRC server."""
def __init__(self, host: str, port: int, tls: bool = True,
tls_verify: bool = True) -> None:
tls_verify: bool = True, proxy: bool = False) -> None:
self.host = host
self.port = port
self.tls = tls
self.tls_verify = tls_verify
self.proxy = proxy
self._reader: asyncio.StreamReader | None = None
self._writer: asyncio.StreamWriter | None = None
@@ -154,10 +155,26 @@ class IRCConnection:
ssl_ctx.check_hostname = False
ssl_ctx.verify_mode = ssl.CERT_NONE
log.info("connecting to %s:%d (tls=%s)", self.host, self.port, self.tls)
self._reader, self._writer = await asyncio.open_connection(
self.host, self.port, ssl=ssl_ctx
)
log.info("connecting to %s:%d (tls=%s, proxy=%s)",
self.host, self.port, self.tls, self.proxy)
if self.proxy:
from derp import http
reader, writer = await http.open_connection(
self.host, self.port,
)
if self.tls:
hostname = self.host if self.tls_verify else None
self._reader, self._writer = await asyncio.open_connection(
sock=writer.transport.get_extra_info("socket"),
ssl=ssl_ctx,
server_hostname=hostname,
)
else:
self._reader, self._writer = reader, writer
else:
self._reader, self._writer = await asyncio.open_connection(
self.host, self.port, ssl=ssl_ctx,
)
log.info("connected")
async def send(self, line: str) -> None:

11
src/derp/mumble.py
View File

@@ -278,7 +278,8 @@ class MumbleBot:
Exposes the same public API as :class:`derp.bot.Bot` so that
protocol-agnostic plugins work without modification.
All TCP goes through ``derp.http.create_connection`` (SOCKS5 proxy).
TCP is routed through ``derp.http.create_connection`` (SOCKS5
optional via ``mumble.proxy`` config).
"""
def __init__(self, name: str, config: dict, registry: PluginRegistry) -> None:
@@ -288,6 +289,7 @@ class MumbleBot:
self._pstate: dict = {}
mu_cfg = config.get("mumble", {})
self._proxy: bool = mu_cfg.get("proxy", True)
self._host: str = mu_cfg.get("host", "127.0.0.1")
self._port: int = mu_cfg.get("port", 64738)
self._username: str = mu_cfg.get("username", "derp")
@@ -332,11 +334,12 @@ class MumbleBot:
return ctx
async def _connect(self) -> None:
"""Establish TLS connection over SOCKS5 proxy."""
"""Establish TLS connection, optionally through SOCKS5 proxy."""
loop = asyncio.get_running_loop()
sock = await loop.run_in_executor(
None, http.create_connection,
(self._host, self._port),
None, lambda: http.create_connection(
(self._host, self._port), proxy=self._proxy,
),
)
ssl_ctx = self._create_ssl_context()
hostname = self._host if self._tls_verify else None

5
src/derp/teams.py
View File

@@ -143,6 +143,7 @@ class TeamsBot:
self._pstate: dict = {}
teams_cfg = config.get("teams", {})
self._proxy: bool = teams_cfg.get("proxy", True)
self.nick: str = teams_cfg.get("bot_name", "derp")
self.prefix: str = config.get("bot", {}).get("prefix", "!")
self._running = False
@@ -390,7 +391,9 @@ class TeamsBot:
)
loop = asyncio.get_running_loop()
try:
await loop.run_in_executor(None, http.urlopen, req)
await loop.run_in_executor(
None, lambda: http.urlopen(req, proxy=self._proxy),
)
except Exception:
log.exception("teams: failed to send via incoming webhook")

6
src/derp/telegram.py
View File

@@ -132,7 +132,8 @@ class TelegramBot:
Exposes the same public API as :class:`derp.bot.Bot` so that
protocol-agnostic plugins work without modification.
All HTTP goes through ``derp.http.urlopen`` (SOCKS5 proxy).
HTTP is routed through ``derp.http.urlopen`` (SOCKS5 optional
via ``telegram.proxy`` config).
"""
def __init__(self, name: str, config: dict, registry: PluginRegistry) -> None:
@@ -144,6 +145,7 @@ class TelegramBot:
tg_cfg = config.get("telegram", {})
self._token: str = tg_cfg.get("bot_token", "")
self._poll_timeout: int = tg_cfg.get("poll_timeout", 30)
self._proxy: bool = tg_cfg.get("proxy", True)
self.nick: str = "" # set by getMe
self._bot_username: str = "" # set by getMe
self.prefix: str = (
@@ -188,7 +190,7 @@ class TelegramBot:
req = urllib.request.Request(url, method="GET")
timeout = self._poll_timeout + 5 if method == "getUpdates" else 30
resp = http.urlopen(req, timeout=timeout)
resp = http.urlopen(req, timeout=timeout, proxy=self._proxy)
body = resp.read() if hasattr(resp, "read") else resp.data
return json.loads(body)

16
tests/test_config.py
View File

@@ -280,3 +280,19 @@ class TestBuildServerConfigs:
assert len(result) == 1
name = list(result.keys())[0]
assert result[name] is raw
class TestProxyDefaults:
"""Verify proxy defaults in each adapter section."""
def test_server_proxy_default_false(self):
assert DEFAULTS["server"]["proxy"] is False
def test_teams_proxy_default_true(self):
assert DEFAULTS["teams"]["proxy"] is True
def test_telegram_proxy_default_true(self):
assert DEFAULTS["telegram"]["proxy"] is True
def test_mumble_proxy_default_true(self):
assert DEFAULTS["mumble"]["proxy"] is True

106
tests/test_http.py
View File

@@ -1,5 +1,6 @@
"""Tests for the SOCKS5 proxy HTTP/TCP module."""
"""Tests for the HTTP/TCP module with optional SOCKS5 proxy."""
import socket
import ssl
import urllib.error
import urllib.request
@@ -267,3 +268,106 @@ class TestCreateConnection:
mock_cls.return_value = sock
result = create_connection(("example.com", 443))
assert result is sock
# -- proxy=False paths -------------------------------------------------------
class TestUrlopenDirect:
"""Tests for urlopen(proxy=False) -- stdlib direct path."""
@patch("derp.http.urllib.request.urlopen")
def test_uses_stdlib_urlopen(self, mock_urlopen):
resp = MagicMock()
mock_urlopen.return_value = resp
result = urlopen("https://example.com/", proxy=False)
mock_urlopen.assert_called_once()
assert result is resp
@patch("derp.http.urllib.request.urlopen")
def test_passes_timeout(self, mock_urlopen):
resp = MagicMock()
mock_urlopen.return_value = resp
urlopen("https://example.com/", timeout=15, proxy=False)
_, kwargs = mock_urlopen.call_args
assert kwargs["timeout"] == 15
@patch("derp.http.urllib.request.urlopen")
def test_passes_context(self, mock_urlopen):
resp = MagicMock()
mock_urlopen.return_value = resp
ctx = ssl.create_default_context()
urlopen("https://example.com/", context=ctx, proxy=False)
_, kwargs = mock_urlopen.call_args
assert kwargs["context"] is ctx
@patch.object(derp.http, "_get_pool")
@patch("derp.http.urllib.request.urlopen")
def test_skips_socks_pool(self, mock_urlopen, mock_pool_fn):
resp = MagicMock()
mock_urlopen.return_value = resp
urlopen("https://example.com/", proxy=False)
mock_pool_fn.assert_not_called()
class TestBuildOpenerDirect:
"""Tests for build_opener(proxy=False) -- no SOCKS5 handler."""
def test_no_proxy_handler(self):
opener = build_opener(proxy=False)
proxy_handlers = [h for h in opener.handlers
if isinstance(h, _ProxyHandler)]
assert len(proxy_handlers) == 0
def test_with_extra_handler(self):
class Custom(urllib.request.HTTPRedirectHandler):
pass
opener = build_opener(Custom, proxy=False)
custom = [h for h in opener.handlers if isinstance(h, Custom)]
assert len(custom) == 1
proxy_handlers = [h for h in opener.handlers
if isinstance(h, _ProxyHandler)]
assert len(proxy_handlers) == 0
class TestCreateConnectionDirect:
"""Tests for create_connection(proxy=False) -- stdlib socket."""
@patch("derp.http.socket.socket")
def test_uses_stdlib_socket(self, mock_sock_cls):
sock = MagicMock()
mock_sock_cls.return_value = sock
result = create_connection(("example.com", 443), proxy=False)
mock_sock_cls.assert_called_once_with(
socket.AF_INET, socket.SOCK_STREAM,
)
assert result is sock
@patch("derp.http.socket.socket")
def test_connects_to_target(self, mock_sock_cls):
sock = MagicMock()
mock_sock_cls.return_value = sock
create_connection(("example.com", 443), proxy=False)
sock.connect.assert_called_once_with(("example.com", 443))
@patch("derp.http.socket.socket")
def test_sets_timeout(self, mock_sock_cls):
sock = MagicMock()
mock_sock_cls.return_value = sock
create_connection(("example.com", 443), timeout=10, proxy=False)
sock.settimeout.assert_called_once_with(10)
@patch("derp.http.socket.socket")
def test_no_socks_proxy_set(self, mock_sock_cls):
sock = MagicMock()
mock_sock_cls.return_value = sock
create_connection(("example.com", 443), proxy=False)
sock.set_proxy.assert_not_called()
@patch("derp.http.socks.socksocket")
@patch("derp.http.socket.socket")
def test_no_socksocket_created(self, mock_sock_cls, mock_socks_cls):
sock = MagicMock()
mock_sock_cls.return_value = sock
create_connection(("example.com", 443), proxy=False)
mock_socks_cls.assert_not_called()

24
tests/test_irc.py
View File

@@ -1,6 +1,12 @@
"""Tests for IRC message parsing and formatting."""
from derp.irc import _parse_tags, _unescape_tag_value, format_msg, parse
from derp.irc import (
IRCConnection,
_parse_tags,
_unescape_tag_value,
format_msg,
parse,
)
class TestParse:
@@ -142,3 +148,19 @@ class TestFormat:
# No space in tail, not starting with colon, head exists -> no colon
result = format_msg("MODE", "#ch", "+o", "nick")
assert result == "MODE #ch +o nick"
class TestIRCConnectionProxy:
"""IRCConnection proxy flag tests."""
def test_proxy_default_false(self):
conn = IRCConnection("irc.example.com", 6697)
assert conn.proxy is False
def test_proxy_enabled(self):
conn = IRCConnection("irc.example.com", 6697, proxy=True)
assert conn.proxy is True
def test_proxy_disabled(self):
conn = IRCConnection("irc.example.com", 6697, proxy=False)
assert conn.proxy is False

23
tests/test_mumble.py
View File

@@ -889,3 +889,26 @@ class TestMumbleBotConfig:
def test_nick_from_username(self):
bot = _make_bot()
assert bot.nick == "derp"
def test_proxy_default_true(self):
bot = _make_bot()
assert bot._proxy is True
def test_proxy_disabled(self):
config = {
"mumble": {
"enabled": True,
"host": "127.0.0.1",
"port": 64738,
"username": "derp",
"password": "",
"tls_verify": False,
"proxy": False,
"admins": [],
"operators": [],
"trusted": [],
},
"bot": {"prefix": "!", "rate_limit": 2.0, "rate_burst": 5},
}
bot = MumbleBot("test", config, PluginRegistry())
assert bot._proxy is False

25
tests/test_teams.py
View File

@@ -732,3 +732,28 @@ class TestTeamsBotPluginManagement:
ok, msg = bot.reload_plugin("nonexistent")
assert ok is False
assert "not loaded" in msg
class TestTeamsBotConfig:
def test_proxy_default_true(self):
bot = _make_bot()
assert bot._proxy is True
def test_proxy_disabled(self):
config = {
"teams": {
"enabled": True,
"bot_name": "derp",
"bind": "127.0.0.1",
"port": 8081,
"webhook_secret": "",
"incoming_webhook_url": "",
"proxy": False,
"admins": [],
"operators": [],
"trusted": [],
},
"bot": {"prefix": "!", "rate_limit": 2.0, "rate_burst": 5},
}
bot = TeamsBot("test", config, PluginRegistry())
assert bot._proxy is False

20
tests/test_telegram.py
View File

@@ -764,3 +764,23 @@ class TestTelegramBotConfig:
def test_admins_coerced_to_str(self):
bot = _make_bot(admins=[111, 222])
assert bot._admins == ["111", "222"]
def test_proxy_default_true(self):
bot = _make_bot()
assert bot._proxy is True
def test_proxy_disabled(self):
config = {
"telegram": {
"enabled": True,
"bot_token": "t",
"poll_timeout": 1,
"proxy": False,
"admins": [],
"operators": [],
"trusted": [],
},
"bot": {"prefix": "!", "rate_limit": 2.0, "rate_burst": 5},
}
bot = TelegramBot("test", config, PluginRegistry())
assert bot._proxy is False