test: add channel filter and JSON log tests
TestChannelFilter: allowed/denied/PM/no-config/core-exempt/ampersand. TestChannelConfig: TOML loading, defaults. TestJsonFormatter: fields, exception, unicode, single-line, timestamp format.
This commit is contained in:
@@ -70,6 +70,25 @@ class TestLoad:
|
|||||||
assert result["bot"]["channels"] == ["#test"]
|
assert result["bot"]["channels"] == ["#test"]
|
||||||
|
|
||||||
|
|
||||||
|
class TestChannelConfig:
|
||||||
|
"""Test channel-level configuration merging."""
|
||||||
|
|
||||||
|
def test_channel_config_loaded(self, tmp_path: Path):
|
||||||
|
config_file = tmp_path / "test.toml"
|
||||||
|
config_file.write_text(
|
||||||
|
'[channels."#ops"]\nplugins = ["core", "dns"]\n'
|
||||||
|
)
|
||||||
|
result = load(config_file)
|
||||||
|
assert "#ops" in result["channels"]
|
||||||
|
assert result["channels"]["#ops"]["plugins"] == ["core", "dns"]
|
||||||
|
|
||||||
|
def test_default_channels_empty(self):
|
||||||
|
assert DEFAULTS["channels"] == {}
|
||||||
|
|
||||||
|
def test_default_logging_format(self):
|
||||||
|
assert DEFAULTS["logging"]["format"] == "text"
|
||||||
|
|
||||||
|
|
||||||
class TestResolveConfig:
|
class TestResolveConfig:
|
||||||
"""Test config path resolution and fallback."""
|
"""Test config path resolution and fallback."""
|
||||||
|
|
||||||
|
|||||||
77
tests/test_log.py
Normal file
77
tests/test_log.py
Normal file
@@ -0,0 +1,77 @@
|
|||||||
|
"""Tests for structured JSON logging."""
|
||||||
|
|
||||||
|
import json
|
||||||
|
import logging
|
||||||
|
|
||||||
|
from derp.log import JsonFormatter
|
||||||
|
|
||||||
|
|
||||||
|
class TestJsonFormatter:
|
||||||
|
"""Test JsonFormatter output."""
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _make_record(msg: str, level: int = logging.INFO,
|
||||||
|
name: str = "derp.test") -> logging.LogRecord:
|
||||||
|
"""Create a LogRecord for testing."""
|
||||||
|
return logging.LogRecord(
|
||||||
|
name=name, level=level, pathname="", lineno=0,
|
||||||
|
msg=msg, args=(), exc_info=None,
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_basic_fields(self):
|
||||||
|
fmt = JsonFormatter()
|
||||||
|
record = self._make_record("hello world")
|
||||||
|
output = fmt.format(record)
|
||||||
|
entry = json.loads(output)
|
||||||
|
assert entry["msg"] == "hello world"
|
||||||
|
assert entry["level"] == "info"
|
||||||
|
assert entry["logger"] == "derp.test"
|
||||||
|
assert "ts" in entry
|
||||||
|
|
||||||
|
def test_level_name(self):
|
||||||
|
fmt = JsonFormatter()
|
||||||
|
record = self._make_record("warn", level=logging.WARNING)
|
||||||
|
entry = json.loads(fmt.format(record))
|
||||||
|
assert entry["level"] == "warning"
|
||||||
|
|
||||||
|
def test_no_exc_key_without_exception(self):
|
||||||
|
fmt = JsonFormatter()
|
||||||
|
record = self._make_record("clean")
|
||||||
|
entry = json.loads(fmt.format(record))
|
||||||
|
assert "exc" not in entry
|
||||||
|
|
||||||
|
def test_exception_included(self):
|
||||||
|
fmt = JsonFormatter()
|
||||||
|
import sys
|
||||||
|
try:
|
||||||
|
raise ValueError("boom")
|
||||||
|
except ValueError:
|
||||||
|
exc_info = sys.exc_info()
|
||||||
|
record = logging.LogRecord(
|
||||||
|
name="derp.test", level=logging.ERROR, pathname="", lineno=0,
|
||||||
|
msg="failed", args=(), exc_info=exc_info,
|
||||||
|
)
|
||||||
|
entry = json.loads(fmt.format(record))
|
||||||
|
assert "exc" in entry
|
||||||
|
assert "ValueError" in entry["exc"]
|
||||||
|
assert "boom" in entry["exc"]
|
||||||
|
|
||||||
|
def test_output_is_single_line(self):
|
||||||
|
fmt = JsonFormatter()
|
||||||
|
record = self._make_record("one liner")
|
||||||
|
output = fmt.format(record)
|
||||||
|
assert "\n" not in output
|
||||||
|
|
||||||
|
def test_unicode_preserved(self):
|
||||||
|
fmt = JsonFormatter()
|
||||||
|
record = self._make_record("cafe\u0301")
|
||||||
|
entry = json.loads(fmt.format(record))
|
||||||
|
assert entry["msg"] == "cafe\u0301"
|
||||||
|
|
||||||
|
def test_timestamp_format(self):
|
||||||
|
fmt = JsonFormatter()
|
||||||
|
record = self._make_record("ts check")
|
||||||
|
entry = json.loads(fmt.format(record))
|
||||||
|
# ISO-ish: YYYY-MM-DDTHH:MM:SS
|
||||||
|
assert "T" in entry["ts"]
|
||||||
|
assert len(entry["ts"]) == 19
|
||||||
@@ -625,6 +625,58 @@ class TestBotAPI:
|
|||||||
assert bot.conn.sent == ["TOPIC #ch :new topic"]
|
assert bot.conn.sent == ["TOPIC #ch :new topic"]
|
||||||
|
|
||||||
|
|
||||||
|
class TestChannelFilter:
|
||||||
|
"""Test per-channel plugin allow/deny."""
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _make_bot(channels_cfg: dict | None = None) -> Bot:
|
||||||
|
"""Create a Bot with optional per-channel config."""
|
||||||
|
config = {
|
||||||
|
"server": {"host": "localhost", "port": 6667, "tls": False,
|
||||||
|
"nick": "test", "user": "test", "realname": "test"},
|
||||||
|
"bot": {"prefix": "!", "channels": [], "plugins_dir": "plugins"},
|
||||||
|
"channels": channels_cfg or {},
|
||||||
|
}
|
||||||
|
return Bot(config, PluginRegistry())
|
||||||
|
|
||||||
|
def test_core_always_allowed(self):
|
||||||
|
bot = self._make_bot({"#locked": {"plugins": ["core"]}})
|
||||||
|
assert bot._plugin_allowed("core", "#locked") is True
|
||||||
|
|
||||||
|
def test_listed_plugin_allowed(self):
|
||||||
|
bot = self._make_bot({"#ops": {"plugins": ["core", "dns"]}})
|
||||||
|
assert bot._plugin_allowed("dns", "#ops") is True
|
||||||
|
|
||||||
|
def test_unlisted_plugin_denied(self):
|
||||||
|
bot = self._make_bot({"#ops": {"plugins": ["core", "dns"]}})
|
||||||
|
assert bot._plugin_allowed("encode", "#ops") is False
|
||||||
|
|
||||||
|
def test_unconfigured_channel_allows_all(self):
|
||||||
|
bot = self._make_bot({"#locked": {"plugins": ["core"]}})
|
||||||
|
assert bot._plugin_allowed("encode", "#open") is True
|
||||||
|
|
||||||
|
def test_no_channels_config_allows_all(self):
|
||||||
|
bot = self._make_bot()
|
||||||
|
assert bot._plugin_allowed("anything", "#test") is True
|
||||||
|
|
||||||
|
def test_pm_always_allowed(self):
|
||||||
|
bot = self._make_bot({"#locked": {"plugins": ["core"]}})
|
||||||
|
assert bot._plugin_allowed("encode", "someone") is True
|
||||||
|
|
||||||
|
def test_none_channel_allowed(self):
|
||||||
|
bot = self._make_bot({"#locked": {"plugins": ["core"]}})
|
||||||
|
assert bot._plugin_allowed("encode", None) is True
|
||||||
|
|
||||||
|
def test_channel_without_plugins_key(self):
|
||||||
|
bot = self._make_bot({"#other": {"some_setting": True}})
|
||||||
|
assert bot._plugin_allowed("encode", "#other") is True
|
||||||
|
|
||||||
|
def test_ampersand_channel(self):
|
||||||
|
bot = self._make_bot({"&local": {"plugins": ["core", "dns"]}})
|
||||||
|
assert bot._plugin_allowed("dns", "&local") is True
|
||||||
|
assert bot._plugin_allowed("encode", "&local") is False
|
||||||
|
|
||||||
|
|
||||||
class TestSplitUtf8:
|
class TestSplitUtf8:
|
||||||
"""Test UTF-8 safe message splitting."""
|
"""Test UTF-8 safe message splitting."""
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user