Add TOML content loader for declarative command definitions
Scan content/commands/ for .toml files at startup and register them as commands alongside Python-defined ones. Two flavors: handler-based (points to a Python callable via module:function) and message-based (auto-generates a handler from inline text). Includes example MOTD command, type validation, error logging, and full test coverage.
This commit is contained in:
parent
8f5956df3d
commit
d159a88ca4
GPG key ID:
22B5C6D66A38B06C
4 changed files with 482 additions and 0 deletions
17
content/commands/motd.toml
Normal file
17
content/commands/motd.toml
Normal file
|
|
@ -0,0 +1,17 @@
|
|||
name = "motd"
|
||||
aliases = ["message"]
|
||||
help = "display the message of the day"
|
||||
mode = "*"
|
||||
message = """
|
||||
=== Message of the Day ===
|
||||
|
||||
Welcome to the MUD!
|
||||
|
||||
This is a procedurally generated world with seamless terrain wrapping.
|
||||
Use cardinal directions (n, s, e, w) or diagonals (ne, nw, se, sw) to move.
|
||||
|
||||
Type 'help' for a list of commands.
|
||||
Type 'fly' to toggle flying mode and soar over mountains and water.
|
||||
|
||||
Have fun exploring!
|
||||
"""
|
||||
141
src/mudlib/content.py
Normal file
141
src/mudlib/content.py
Normal file
|
|
@ -0,0 +1,141 @@
|
|||
"""Content loading from TOML files."""
|
||||
|
||||
import importlib
|
||||
import logging
|
||||
import tomllib
|
||||
from pathlib import Path
|
||||
from typing import cast
|
||||
|
||||
from mudlib.commands import CommandDefinition, CommandHandler
|
||||
from mudlib.player import Player
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _resolve_handler(handler_ref: str) -> CommandHandler:
|
||||
"""Resolve a handler reference string to a callable.
|
||||
|
||||
Args:
|
||||
handler_ref: String in format "module.path:function_name"
|
||||
|
||||
Returns:
|
||||
The resolved callable
|
||||
|
||||
Raises:
|
||||
ImportError: If the module cannot be imported
|
||||
AttributeError: If the function doesn't exist in the module
|
||||
"""
|
||||
if ":" not in handler_ref:
|
||||
msg = (
|
||||
f"Handler reference must be in format 'module:function', got: {handler_ref}"
|
||||
)
|
||||
raise ValueError(msg)
|
||||
|
||||
module_path, func_name = handler_ref.split(":", 1)
|
||||
module = importlib.import_module(module_path)
|
||||
handler = getattr(module, func_name)
|
||||
|
||||
if not callable(handler):
|
||||
msg = f"Handler reference {handler_ref} is not callable"
|
||||
raise TypeError(msg)
|
||||
|
||||
return cast(CommandHandler, handler)
|
||||
|
||||
|
||||
def _make_message_handler(message: str) -> CommandHandler:
|
||||
"""Create a handler that sends a fixed message to the player.
|
||||
|
||||
Args:
|
||||
message: The message text to send
|
||||
|
||||
Returns:
|
||||
An async handler function
|
||||
"""
|
||||
|
||||
async def handler(player: Player, args: str) -> None:
|
||||
player.writer.write(message.strip() + "\r\n")
|
||||
await player.writer.drain()
|
||||
|
||||
return handler
|
||||
|
||||
|
||||
def load_command(path: Path) -> CommandDefinition:
|
||||
"""Load a command definition from a TOML file.
|
||||
|
||||
Args:
|
||||
path: Path to the .toml file
|
||||
|
||||
Returns:
|
||||
CommandDefinition instance
|
||||
|
||||
Raises:
|
||||
KeyError: If required fields are missing
|
||||
ValueError: If neither handler nor message is specified
|
||||
TypeError: If handler or message fields have wrong types
|
||||
ImportError: If handler reference cannot be resolved
|
||||
"""
|
||||
with open(path, "rb") as f:
|
||||
data = tomllib.load(f)
|
||||
|
||||
name = data["name"]
|
||||
|
||||
# Determine the handler
|
||||
handler_ref = data.get("handler")
|
||||
message = data.get("message")
|
||||
|
||||
if handler_ref and message:
|
||||
msg = "Command definition cannot specify both 'handler' and 'message'"
|
||||
raise ValueError(msg)
|
||||
|
||||
if not handler_ref and not message:
|
||||
msg = "Command definition must specify either 'handler' or 'message'"
|
||||
raise ValueError(msg)
|
||||
|
||||
# Validate types
|
||||
if handler_ref and not isinstance(handler_ref, str):
|
||||
msg = f"'handler' must be a string in {path}"
|
||||
raise TypeError(msg)
|
||||
|
||||
if message and not isinstance(message, str):
|
||||
msg = f"'message' must be a string in {path}"
|
||||
raise TypeError(msg)
|
||||
|
||||
handler: CommandHandler
|
||||
if handler_ref:
|
||||
handler = _resolve_handler(handler_ref)
|
||||
else:
|
||||
# message must be a string here since we checked it's not None
|
||||
handler = _make_message_handler(cast(str, message))
|
||||
|
||||
# Optional fields with defaults
|
||||
aliases = data.get("aliases", [])
|
||||
mode = data.get("mode", "normal")
|
||||
help_text = data.get("help", "")
|
||||
|
||||
return CommandDefinition(
|
||||
name=name,
|
||||
handler=handler,
|
||||
aliases=aliases,
|
||||
mode=mode,
|
||||
help=help_text,
|
||||
)
|
||||
|
||||
|
||||
def load_commands(directory: Path) -> list[CommandDefinition]:
|
||||
"""Load all command definitions from a directory.
|
||||
|
||||
Args:
|
||||
directory: Path to directory containing .toml files
|
||||
|
||||
Returns:
|
||||
List of CommandDefinition instances
|
||||
"""
|
||||
commands = []
|
||||
|
||||
for path in directory.glob("*.toml"):
|
||||
try:
|
||||
commands.append(load_command(path))
|
||||
except Exception as e:
|
||||
log.warning("failed to load command from %s: %s", path, e)
|
||||
|
||||
return commands
|
||||
|
|
@ -15,6 +15,7 @@ import mudlib.commands.fly
|
|||
import mudlib.commands.look
|
||||
import mudlib.commands.movement
|
||||
import mudlib.commands.quit
|
||||
from mudlib.content import load_commands
|
||||
from mudlib.effects import clear_expired
|
||||
from mudlib.player import Player, players
|
||||
from mudlib.world.terrain import World
|
||||
|
|
@ -189,6 +190,15 @@ async def run_server() -> None:
|
|||
mudlib.commands.look.world = _world
|
||||
mudlib.commands.movement.world = _world
|
||||
|
||||
# Load content-defined commands from TOML files
|
||||
content_dir = pathlib.Path(__file__).resolve().parents[2] / "content" / "commands"
|
||||
if content_dir.exists():
|
||||
log.info("loading content from %s", content_dir)
|
||||
content_commands = load_commands(content_dir)
|
||||
for cmd_def in content_commands:
|
||||
mudlib.commands.register(cmd_def)
|
||||
log.debug("registered content command: %s", cmd_def.name)
|
||||
|
||||
# connect_maxwait: how long to wait for telnet option negotiation (CHARSET
|
||||
# etc) before starting the shell. default is 4.0s which is painful.
|
||||
# MUD clients like tintin++ reject CHARSET immediately via MTTS, but
|
||||
|
|
|
|||
314
tests/test_content_loader.py
Normal file
314
tests/test_content_loader.py
Normal file
|
|
@ -0,0 +1,314 @@
|
|||
"""Tests for content loading from TOML files."""
|
||||
|
||||
import logging
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from mudlib.content import load_command, load_commands
|
||||
from mudlib.player import Player
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_writer():
|
||||
writer = MagicMock()
|
||||
writer.write = MagicMock()
|
||||
writer.drain = AsyncMock()
|
||||
return writer
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_reader():
|
||||
return MagicMock()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def player(mock_reader, mock_writer):
|
||||
return Player(name="TestPlayer", x=5, y=5, reader=mock_reader, writer=mock_writer)
|
||||
|
||||
|
||||
def test_load_command_with_handler_reference():
|
||||
"""Test loading a command with a Python handler reference."""
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".toml", delete=False) as f:
|
||||
f.write("""
|
||||
name = "shout"
|
||||
aliases = ["yell"]
|
||||
help = "shout a message to nearby players"
|
||||
mode = "normal"
|
||||
handler = "mudlib.commands.look:cmd_look"
|
||||
""")
|
||||
f.flush()
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
defn = load_command(tmp_path)
|
||||
assert defn.name == "shout"
|
||||
assert defn.aliases == ["yell"]
|
||||
assert defn.help == "shout a message to nearby players"
|
||||
assert defn.mode == "normal"
|
||||
assert callable(defn.handler)
|
||||
finally:
|
||||
tmp_path.unlink()
|
||||
|
||||
|
||||
def test_load_command_with_message():
|
||||
"""Test loading a command that just sends a message."""
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".toml", delete=False) as f:
|
||||
f.write("""
|
||||
name = "rules"
|
||||
aliases = ["info"]
|
||||
help = "display the server rules"
|
||||
mode = "*"
|
||||
message = \"\"\"
|
||||
Welcome to the MUD!
|
||||
1. Be respectful
|
||||
2. No cheating
|
||||
\"\"\"
|
||||
""")
|
||||
f.flush()
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
defn = load_command(tmp_path)
|
||||
assert defn.name == "rules"
|
||||
assert defn.aliases == ["info"]
|
||||
assert defn.help == "display the server rules"
|
||||
assert defn.mode == "*"
|
||||
assert callable(defn.handler)
|
||||
finally:
|
||||
tmp_path.unlink()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_message_command_sends_text(player, mock_writer):
|
||||
"""Test that a message command actually sends the text to the player."""
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".toml", delete=False) as f:
|
||||
f.write("""
|
||||
name = "motd"
|
||||
message = "Welcome to the server!\\nEnjoy your stay."
|
||||
""")
|
||||
f.flush()
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
defn = load_command(tmp_path)
|
||||
await defn.handler(player, "")
|
||||
|
||||
assert mock_writer.write.called
|
||||
written_text = mock_writer.write.call_args[0][0]
|
||||
assert "Welcome to the server!" in written_text
|
||||
assert "Enjoy your stay." in written_text
|
||||
finally:
|
||||
tmp_path.unlink()
|
||||
|
||||
|
||||
def test_load_commands_from_directory():
|
||||
"""Test loading multiple command definitions from a directory."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tmp_path = Path(tmpdir)
|
||||
|
||||
# Create two command files
|
||||
(tmp_path / "cmd1.toml").write_text("""
|
||||
name = "cmd1"
|
||||
message = "Command 1"
|
||||
""")
|
||||
(tmp_path / "cmd2.toml").write_text("""
|
||||
name = "cmd2"
|
||||
message = "Command 2"
|
||||
""")
|
||||
|
||||
# Create a non-toml file that should be ignored
|
||||
(tmp_path / "readme.txt").write_text("This is not a command")
|
||||
|
||||
commands = load_commands(tmp_path)
|
||||
|
||||
assert len(commands) == 2
|
||||
names = {cmd.name for cmd in commands}
|
||||
assert names == {"cmd1", "cmd2"}
|
||||
|
||||
|
||||
def test_load_command_missing_name():
|
||||
"""Test that missing name field raises an error."""
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".toml", delete=False) as f:
|
||||
f.write("""
|
||||
message = "Hello"
|
||||
""")
|
||||
f.flush()
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
with pytest.raises(KeyError):
|
||||
load_command(tmp_path)
|
||||
finally:
|
||||
tmp_path.unlink()
|
||||
|
||||
|
||||
def test_load_command_missing_handler_and_message():
|
||||
"""Test that a command without handler or message raises an error."""
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".toml", delete=False) as f:
|
||||
f.write("""
|
||||
name = "broken"
|
||||
""")
|
||||
f.flush()
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
match_msg = "must specify either 'handler' or 'message'"
|
||||
with pytest.raises(ValueError, match=match_msg):
|
||||
load_command(tmp_path)
|
||||
finally:
|
||||
tmp_path.unlink()
|
||||
|
||||
|
||||
def test_load_command_bad_handler_reference():
|
||||
"""Test that an invalid handler reference raises an error."""
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".toml", delete=False) as f:
|
||||
f.write("""
|
||||
name = "broken"
|
||||
handler = "nonexistent.module:function"
|
||||
""")
|
||||
f.flush()
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
with pytest.raises((ImportError, ModuleNotFoundError)):
|
||||
load_command(tmp_path)
|
||||
finally:
|
||||
tmp_path.unlink()
|
||||
|
||||
|
||||
def test_load_command_defaults():
|
||||
"""Test that optional fields have sensible defaults."""
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".toml", delete=False) as f:
|
||||
f.write("""
|
||||
name = "simple"
|
||||
message = "Hello"
|
||||
""")
|
||||
f.flush()
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
defn = load_command(tmp_path)
|
||||
assert defn.name == "simple"
|
||||
assert defn.aliases == []
|
||||
assert defn.help == ""
|
||||
assert defn.mode == "normal"
|
||||
finally:
|
||||
tmp_path.unlink()
|
||||
|
||||
|
||||
def test_load_command_both_handler_and_message():
|
||||
"""Test that specifying both handler and message raises an error."""
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".toml", delete=False) as f:
|
||||
f.write("""
|
||||
name = "broken"
|
||||
handler = "mudlib.commands.look:cmd_look"
|
||||
message = "Hello"
|
||||
""")
|
||||
f.flush()
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
match_msg = "cannot specify both 'handler' and 'message'"
|
||||
with pytest.raises(ValueError, match=match_msg):
|
||||
load_command(tmp_path)
|
||||
finally:
|
||||
tmp_path.unlink()
|
||||
|
||||
|
||||
def test_load_command_handler_wrong_type():
|
||||
"""Test that non-string handler field raises TypeError."""
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".toml", delete=False) as f:
|
||||
f.write("""
|
||||
name = "broken"
|
||||
handler = 123
|
||||
""")
|
||||
f.flush()
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
match_msg = "'handler' must be a string"
|
||||
with pytest.raises(TypeError, match=match_msg):
|
||||
load_command(tmp_path)
|
||||
finally:
|
||||
tmp_path.unlink()
|
||||
|
||||
|
||||
def test_load_command_message_wrong_type():
|
||||
"""Test that non-string message field raises TypeError."""
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".toml", delete=False) as f:
|
||||
f.write("""
|
||||
name = "broken"
|
||||
message = 456
|
||||
""")
|
||||
f.flush()
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
match_msg = "'message' must be a string"
|
||||
with pytest.raises(TypeError, match=match_msg):
|
||||
load_command(tmp_path)
|
||||
finally:
|
||||
tmp_path.unlink()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_message_whitespace_handling(player, mock_writer):
|
||||
"""Test message stripping removes outer whitespace, keeps internal formatting."""
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".toml", delete=False) as f:
|
||||
f.write("""
|
||||
name = "formatted"
|
||||
message = \"\"\"
|
||||
|
||||
First line
|
||||
Indented line
|
||||
Last line
|
||||
|
||||
\"\"\"
|
||||
""")
|
||||
f.flush()
|
||||
tmp_path = Path(f.name)
|
||||
|
||||
try:
|
||||
defn = load_command(tmp_path)
|
||||
await defn.handler(player, "")
|
||||
|
||||
assert mock_writer.write.called
|
||||
written_text = mock_writer.write.call_args[0][0]
|
||||
# Outer whitespace should be stripped
|
||||
assert not written_text.startswith("\n")
|
||||
# But internal formatting should be preserved
|
||||
assert "First line\n Indented line\nLast line" in written_text
|
||||
finally:
|
||||
tmp_path.unlink()
|
||||
|
||||
|
||||
def test_load_commands_logs_broken_files(caplog):
|
||||
"""Test that load_commands logs warnings for broken TOML files."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tmp_path = Path(tmpdir)
|
||||
|
||||
# Create a valid command file
|
||||
(tmp_path / "valid.toml").write_text("""
|
||||
name = "valid"
|
||||
message = "OK"
|
||||
""")
|
||||
|
||||
# Create a broken command file (missing required field)
|
||||
(tmp_path / "broken.toml").write_text("""
|
||||
name = "broken"
|
||||
""")
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
commands = load_commands(tmp_path)
|
||||
|
||||
# Should load the valid one only
|
||||
assert len(commands) == 1
|
||||
assert commands[0].name == "valid"
|
||||
|
||||
# Should log warning about the broken file
|
||||
warnings = [r.message for r in caplog.records]
|
||||
assert any("failed to load command from" in msg for msg in warnings)
|
||||
assert any("broken.toml" in msg for msg in warnings)
|
||||
Loading…
Reference in a new issue