Add player alias system with persistence and dispatch
Implements a complete alias system allowing players to create command shortcuts. Aliases are expanded during dispatch with a recursion guard (max 10 levels). Changes: - Add aliases field to Player dataclass (dict[str, str]) - Add player_aliases table to database schema - Add save_aliases() and load_aliases() persistence functions - Add alias/unalias commands with built-in command protection - Integrate alias expansion into dispatch() before command resolution - Add comprehensive test coverage for all features
This commit is contained in:
parent
4c969d2987
commit
3f042de360
GPG key ID:
22B5C6D66A38B06C
6 changed files with 409 additions and 0 deletions
|
|
@ -141,6 +141,23 @@ async def dispatch(player: Player, raw_input: str) -> None:
|
|||
command = parts[0].lower()
|
||||
args = parts[1] if len(parts) > 1 else ""
|
||||
|
||||
# Resolve aliases (with recursion guard)
|
||||
expansion_count = 0
|
||||
max_expansions = 10
|
||||
while command in player.aliases:
|
||||
if expansion_count >= max_expansions:
|
||||
player.writer.write("Too many nested aliases (max 10).\r\n")
|
||||
await player.writer.drain()
|
||||
return
|
||||
expansion = player.aliases[command]
|
||||
# Combine expansion with remaining args
|
||||
raw_input = f"{expansion} {args}" if args else expansion
|
||||
# Re-split to get new command and args
|
||||
parts = raw_input.split(maxsplit=1)
|
||||
command = parts[0].lower()
|
||||
args = parts[1] if len(parts) > 1 else ""
|
||||
expansion_count += 1
|
||||
|
||||
# Resolve command by exact match or prefix
|
||||
result = resolve_prefix(command)
|
||||
|
||||
|
|
|
|||
74
src/mudlib/commands/alias.py
Normal file
74
src/mudlib/commands/alias.py
Normal file
|
|
@ -0,0 +1,74 @@
|
|||
"""Player alias commands."""
|
||||
|
||||
from mudlib.commands import CommandDefinition, _registry, register
|
||||
from mudlib.player import Player
|
||||
|
||||
|
||||
async def cmd_alias(player: Player, args: str) -> None:
|
||||
"""Create or list aliases.
|
||||
|
||||
Usage:
|
||||
alias - List all aliases
|
||||
alias <name> - Show a specific alias
|
||||
alias <name> <exp> - Create an alias
|
||||
"""
|
||||
args = args.strip()
|
||||
|
||||
# No args: list all aliases
|
||||
if not args:
|
||||
if not player.aliases:
|
||||
await player.send("No aliases defined.\r\n")
|
||||
else:
|
||||
lines = [
|
||||
f"{alias} -> {expansion}"
|
||||
for alias, expansion in sorted(player.aliases.items())
|
||||
]
|
||||
await player.send("\r\n".join(lines) + "\r\n")
|
||||
return
|
||||
|
||||
# Check if this is a single-word lookup or a definition
|
||||
parts = args.split(None, 1)
|
||||
alias_name = parts[0]
|
||||
|
||||
if len(parts) == 1:
|
||||
# Show single alias
|
||||
if alias_name in player.aliases:
|
||||
await player.send(f"{alias_name} -> {player.aliases[alias_name]}\r\n")
|
||||
else:
|
||||
await player.send(f"No such alias: {alias_name}\r\n")
|
||||
return
|
||||
|
||||
# Create alias
|
||||
expansion = parts[1]
|
||||
|
||||
# Cannot alias over built-in commands
|
||||
if alias_name in _registry:
|
||||
await player.send(f"Cannot alias over built-in command: {alias_name}\r\n")
|
||||
return
|
||||
|
||||
player.aliases[alias_name] = expansion
|
||||
await player.send(f"Alias set: {alias_name} -> {expansion}\r\n")
|
||||
|
||||
|
||||
async def cmd_unalias(player: Player, args: str) -> None:
|
||||
"""Remove an alias.
|
||||
|
||||
Usage:
|
||||
unalias <name>
|
||||
"""
|
||||
alias_name = args.strip()
|
||||
|
||||
if not alias_name:
|
||||
await player.send("Usage: unalias <name>\r\n")
|
||||
return
|
||||
|
||||
if alias_name in player.aliases:
|
||||
del player.aliases[alias_name]
|
||||
await player.send(f"Alias removed: {alias_name}\r\n")
|
||||
else:
|
||||
await player.send(f"No such alias: {alias_name}\r\n")
|
||||
|
||||
|
||||
# Register commands
|
||||
register(CommandDefinition("alias", cmd_alias))
|
||||
register(CommandDefinition("unalias", cmd_unalias))
|
||||
|
|
@ -32,6 +32,7 @@ class Player(Entity):
|
|||
painting: bool = False
|
||||
paint_brush: str = "."
|
||||
prompt_template: str | None = None
|
||||
aliases: dict[str, str] = field(default_factory=dict)
|
||||
_last_msdp: dict = field(default_factory=dict, repr=False)
|
||||
_power_task: asyncio.Task | None = None
|
||||
|
||||
|
|
|
|||
|
|
@ -55,6 +55,7 @@ from mudlib.store import (
|
|||
authenticate,
|
||||
create_account,
|
||||
init_db,
|
||||
load_aliases,
|
||||
load_player_data,
|
||||
save_player,
|
||||
update_last_login,
|
||||
|
|
@ -344,6 +345,9 @@ async def shell(
|
|||
reader=_reader,
|
||||
)
|
||||
|
||||
# Load aliases from database
|
||||
player.aliases = load_aliases(player_name)
|
||||
|
||||
# Reconstruct inventory from saved data
|
||||
for item_name in player_data.get("inventory", []):
|
||||
template = thing_templates.get(item_name)
|
||||
|
|
|
|||
|
|
@ -62,6 +62,15 @@ def init_db(db_path: str | Path) -> None:
|
|||
)
|
||||
""")
|
||||
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS player_aliases (
|
||||
player_name TEXT NOT NULL COLLATE NOCASE,
|
||||
alias TEXT NOT NULL COLLATE NOCASE,
|
||||
expansion TEXT NOT NULL,
|
||||
PRIMARY KEY (player_name, alias)
|
||||
)
|
||||
""")
|
||||
|
||||
# Migrations: add columns if they don't exist (old schemas)
|
||||
cursor.execute("PRAGMA table_info(accounts)")
|
||||
columns = [row[1] for row in cursor.fetchall()]
|
||||
|
|
@ -226,6 +235,9 @@ def save_player(player: Player) -> None:
|
|||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
# Save aliases
|
||||
save_aliases(player.name, player.aliases)
|
||||
|
||||
|
||||
def load_player_data(name: str) -> PlayerData | None:
|
||||
"""Load player data from the database.
|
||||
|
|
@ -303,3 +315,59 @@ def update_last_login(name: str) -> None:
|
|||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
|
||||
def save_aliases(
|
||||
name: str, aliases: dict[str, str], db_path: str | Path | None = None
|
||||
) -> None:
|
||||
"""Save player aliases to the database.
|
||||
|
||||
Replaces all existing aliases for the player.
|
||||
|
||||
Args:
|
||||
name: Player name (case-insensitive)
|
||||
aliases: Dictionary mapping alias names to expansions
|
||||
db_path: Optional explicit database path (for testing)
|
||||
"""
|
||||
conn = sqlite3.connect(str(db_path)) if db_path is not None else _get_connection()
|
||||
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Delete all existing aliases for this player
|
||||
cursor.execute("DELETE FROM player_aliases WHERE player_name = ?", (name,))
|
||||
|
||||
# Insert new aliases
|
||||
for alias, expansion in aliases.items():
|
||||
cursor.execute(
|
||||
"INSERT INTO player_aliases (player_name, alias, expansion) "
|
||||
"VALUES (?, ?, ?)",
|
||||
(name, alias, expansion),
|
||||
)
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
|
||||
def load_aliases(name: str, db_path: str | Path | None = None) -> dict[str, str]:
|
||||
"""Load player aliases from the database.
|
||||
|
||||
Args:
|
||||
name: Player name (case-insensitive)
|
||||
db_path: Optional explicit database path (for testing)
|
||||
|
||||
Returns:
|
||||
Dictionary mapping alias names to expansions
|
||||
"""
|
||||
conn = sqlite3.connect(str(db_path)) if db_path is not None else _get_connection()
|
||||
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute(
|
||||
"SELECT alias, expansion FROM player_aliases WHERE player_name = ?",
|
||||
(name,),
|
||||
)
|
||||
|
||||
result = {alias: expansion for alias, expansion in cursor.fetchall()}
|
||||
|
||||
conn.close()
|
||||
return result
|
||||
|
|
|
|||
245
tests/test_alias.py
Normal file
245
tests/test_alias.py
Normal file
|
|
@ -0,0 +1,245 @@
|
|||
"""Tests for player alias system."""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from mudlib.commands import CommandDefinition, dispatch, register
|
||||
from mudlib.player import Player
|
||||
from mudlib.store import (
|
||||
create_account,
|
||||
init_db,
|
||||
load_aliases,
|
||||
save_aliases,
|
||||
save_player,
|
||||
)
|
||||
|
||||
|
||||
# Persistence tests
|
||||
def test_save_and_load_aliases_roundtrip():
|
||||
"""Save and load aliases from database."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
db_path = Path(tmpdir) / "test.db"
|
||||
init_db(db_path)
|
||||
|
||||
aliases = {"pr": "punch right", "pl": "punch left", "l": "look"}
|
||||
save_aliases("goku", aliases, db_path)
|
||||
|
||||
loaded = load_aliases("goku", db_path)
|
||||
assert loaded == aliases
|
||||
|
||||
|
||||
def test_load_aliases_empty():
|
||||
"""Loading aliases for player with none returns empty dict."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
db_path = Path(tmpdir) / "test.db"
|
||||
init_db(db_path)
|
||||
|
||||
loaded = load_aliases("goku", db_path)
|
||||
assert loaded == {}
|
||||
|
||||
|
||||
def test_save_aliases_overwrites_existing():
|
||||
"""Saving new aliases replaces old ones."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
db_path = Path(tmpdir) / "test.db"
|
||||
init_db(db_path)
|
||||
|
||||
save_aliases("goku", {"pr": "punch right"}, db_path)
|
||||
save_aliases("goku", {"pl": "punch left", "l": "look"}, db_path)
|
||||
|
||||
loaded = load_aliases("goku", db_path)
|
||||
assert loaded == {"pl": "punch left", "l": "look"}
|
||||
assert "pr" not in loaded
|
||||
|
||||
|
||||
# Alias command tests
|
||||
@pytest.mark.asyncio
|
||||
async def test_alias_list_empty(player):
|
||||
"""alias with no args shows message when no aliases defined."""
|
||||
from mudlib.commands.alias import cmd_alias
|
||||
|
||||
await cmd_alias(player, "")
|
||||
player.writer.write.assert_called_with("No aliases defined.\r\n")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_alias_create(player):
|
||||
"""alias <name> <expansion> creates an alias."""
|
||||
from mudlib.commands.alias import cmd_alias
|
||||
|
||||
await cmd_alias(player, "pr punch right")
|
||||
assert player.aliases["pr"] == "punch right"
|
||||
player.writer.write.assert_called_with("Alias set: pr -> punch right\r\n")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_alias_list_with_aliases(player):
|
||||
"""alias with no args lists all aliases."""
|
||||
from mudlib.commands.alias import cmd_alias
|
||||
|
||||
player.aliases = {"pr": "punch right", "pl": "punch left", "l": "look"}
|
||||
|
||||
await cmd_alias(player, "")
|
||||
output = player.writer.write.call_args[0][0]
|
||||
assert "pr -> punch right" in output
|
||||
assert "pl -> punch left" in output
|
||||
assert "l -> look" in output
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_alias_show_single(player):
|
||||
"""alias <name> shows that specific alias."""
|
||||
from mudlib.commands.alias import cmd_alias
|
||||
|
||||
player.aliases = {"pr": "punch right", "pl": "punch left"}
|
||||
|
||||
await cmd_alias(player, "pr")
|
||||
player.writer.write.assert_called_with("pr -> punch right\r\n")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unalias_removes_alias(player):
|
||||
"""unalias <name> removes an alias."""
|
||||
from mudlib.commands.alias import cmd_unalias
|
||||
|
||||
player.aliases = {"pr": "punch right"}
|
||||
|
||||
await cmd_unalias(player, "pr")
|
||||
assert "pr" not in player.aliases
|
||||
player.writer.write.assert_called_with("Alias removed: pr\r\n")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unalias_no_such_alias(player):
|
||||
"""unalias on non-existent alias shows error."""
|
||||
from mudlib.commands.alias import cmd_unalias
|
||||
|
||||
await cmd_unalias(player, "pr")
|
||||
player.writer.write.assert_called_with("No such alias: pr\r\n")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_alias_cannot_override_builtin(player):
|
||||
"""Cannot alias over existing built-in commands."""
|
||||
import mudlib.commands.look # noqa: F401 - needed to register look command
|
||||
from mudlib.commands.alias import cmd_alias
|
||||
|
||||
await cmd_alias(player, "look punch right")
|
||||
player.writer.write.assert_called_with(
|
||||
"Cannot alias over built-in command: look\r\n"
|
||||
)
|
||||
assert "look" not in player.aliases
|
||||
|
||||
|
||||
# Dispatch integration tests
|
||||
@pytest.mark.asyncio
|
||||
async def test_alias_expands_in_dispatch(player):
|
||||
"""Aliases are expanded before command dispatch."""
|
||||
called_with = []
|
||||
|
||||
async def test_handler(p, args):
|
||||
called_with.append(args)
|
||||
|
||||
register(CommandDefinition("testcmd", test_handler))
|
||||
player.aliases["tc"] = "testcmd"
|
||||
|
||||
await dispatch(player, "tc hello")
|
||||
assert called_with == ["hello"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_alias_with_extra_args(player):
|
||||
"""Alias expansion preserves additional arguments."""
|
||||
called_with = []
|
||||
|
||||
async def test_handler(p, args):
|
||||
called_with.append(args)
|
||||
|
||||
register(CommandDefinition("testcmd", test_handler))
|
||||
player.aliases["tc"] = "testcmd arg1"
|
||||
|
||||
await dispatch(player, "tc arg2 arg3")
|
||||
# Expansion: "testcmd arg1" + " arg2 arg3" = "testcmd arg1 arg2 arg3"
|
||||
assert called_with == ["arg1 arg2 arg3"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_nested_alias_max_depth(player):
|
||||
"""Nested aliases are limited to prevent infinite recursion."""
|
||||
called = []
|
||||
|
||||
async def test_handler(p, args):
|
||||
called.append(True)
|
||||
|
||||
register(CommandDefinition("final", test_handler))
|
||||
|
||||
# Create a chain: a -> b -> c -> ... -> final
|
||||
player.aliases["a"] = "b"
|
||||
player.aliases["b"] = "c"
|
||||
player.aliases["c"] = "d"
|
||||
player.aliases["d"] = "e"
|
||||
player.aliases["e"] = "f"
|
||||
player.aliases["f"] = "g"
|
||||
player.aliases["g"] = "h"
|
||||
player.aliases["h"] = "i"
|
||||
player.aliases["i"] = "j"
|
||||
player.aliases["j"] = "final"
|
||||
|
||||
await dispatch(player, "a")
|
||||
assert len(called) == 1 # Should complete successfully
|
||||
|
||||
# Now create an 11-deep chain that exceeds limit
|
||||
player.aliases["j"] = "k"
|
||||
player.aliases["k"] = "final"
|
||||
|
||||
called.clear()
|
||||
await dispatch(player, "a")
|
||||
# Should fail and send error message
|
||||
assert len(called) == 0
|
||||
assert "Too many nested aliases" in player.writer.write.call_args[0][0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unknown_alias_falls_through(player):
|
||||
"""Unknown aliases fall through to normal command resolution."""
|
||||
player.aliases = {} # No aliases defined
|
||||
|
||||
# This should hit normal "Unknown command" path
|
||||
await dispatch(player, "nonexistent")
|
||||
assert "Unknown command" in player.writer.write.call_args[0][0]
|
||||
|
||||
|
||||
# Integration test for persistence
|
||||
def test_aliases_persist_on_save_player():
|
||||
"""Aliases are saved when save_player is called."""
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as f:
|
||||
db_path = f.name
|
||||
|
||||
try:
|
||||
init_db(db_path)
|
||||
create_account("Goku", "password")
|
||||
|
||||
# Create player with aliases
|
||||
mock_writer = MagicMock()
|
||||
player = Player(
|
||||
name="Goku",
|
||||
x=10,
|
||||
y=20,
|
||||
reader=MagicMock(),
|
||||
writer=mock_writer,
|
||||
)
|
||||
player.aliases = {"pr": "punch right", "pl": "punch left", "l": "look"}
|
||||
|
||||
# Save player (should save aliases too)
|
||||
save_player(player)
|
||||
|
||||
# Load aliases directly from database
|
||||
loaded = load_aliases("Goku")
|
||||
assert loaded == {"pr": "punch right", "pl": "punch left", "l": "look"}
|
||||
|
||||
finally:
|
||||
os.unlink(db_path)
|
||||
Loading…
Reference in a new issue