Jared Miller
4cff1475c3
Player objects were removed from the players dict on quit/disconnect but never removed from zone._contents, leaving ghost * markers on other players' maps.
472 lines
14 KiB
Python
472 lines
14 KiB
Python
"""Tests for the command system."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from mudlib import commands
|
|
from mudlib.commands import CommandDefinition, look, movement
|
|
from mudlib.effects import active_effects, add_effect
|
|
from mudlib.player import Player
|
|
from mudlib.render.ansi import RESET
|
|
from mudlib.zone import Zone
|
|
|
|
|
|
@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 test_zone():
|
|
# Create a 100x100 zone filled with passable terrain
|
|
terrain = [["." for _ in range(100)] for _ in range(100)]
|
|
zone = Zone(
|
|
name="testzone",
|
|
width=100,
|
|
height=100,
|
|
toroidal=True,
|
|
terrain=terrain,
|
|
impassable=set(), # All terrain is passable for tests
|
|
)
|
|
return zone
|
|
|
|
|
|
@pytest.fixture
|
|
def player(mock_reader, mock_writer, test_zone):
|
|
p = Player(name="TestPlayer", x=5, y=5, reader=mock_reader, writer=mock_writer)
|
|
p.location = test_zone
|
|
test_zone._contents.append(p)
|
|
return p
|
|
|
|
|
|
# Test command registration
|
|
def test_register_command():
|
|
"""Test that commands can be registered."""
|
|
|
|
async def test_handler(player, args):
|
|
pass
|
|
|
|
commands.register(CommandDefinition("test", test_handler))
|
|
assert "test" in commands._registry
|
|
assert commands._registry["test"].handler is test_handler
|
|
|
|
|
|
def test_register_command_with_aliases():
|
|
"""Test that command aliases work."""
|
|
|
|
async def test_handler(player, args):
|
|
pass
|
|
|
|
commands.register(CommandDefinition("testcmd", test_handler, aliases=["tc", "t"]))
|
|
assert "testcmd" in commands._registry
|
|
assert "tc" in commands._registry
|
|
assert "t" in commands._registry
|
|
assert commands._registry["testcmd"] is commands._registry["tc"]
|
|
assert commands._registry["testcmd"] is commands._registry["t"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatch_routes_to_handler(player):
|
|
"""Test that dispatch routes input to the correct handler."""
|
|
called = False
|
|
received_args = None
|
|
|
|
async def test_handler(p, args):
|
|
nonlocal called, received_args
|
|
called = True
|
|
received_args = args
|
|
|
|
commands.register(CommandDefinition("testcmd", test_handler))
|
|
await commands.dispatch(player, "testcmd arg1 arg2")
|
|
|
|
assert called
|
|
assert received_args == "arg1 arg2"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatch_handles_unknown_command(player, mock_writer):
|
|
"""Test that unknown commands give feedback."""
|
|
await commands.dispatch(player, "unknowncommand")
|
|
|
|
# Should have written some kind of error message
|
|
assert mock_writer.write.called
|
|
error_msg = mock_writer.write.call_args[0][0]
|
|
assert "unknown" in error_msg.lower() or "not found" in error_msg.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatch_handles_empty_input(player):
|
|
"""Test that empty input doesn't crash."""
|
|
await commands.dispatch(player, "")
|
|
await commands.dispatch(player, " ")
|
|
|
|
|
|
# Test movement direction parsing
|
|
@pytest.mark.parametrize(
|
|
"direction,expected_delta",
|
|
[
|
|
("n", (0, -1)),
|
|
("north", (0, -1)),
|
|
("s", (0, 1)),
|
|
("south", (0, 1)),
|
|
("e", (1, 0)),
|
|
("east", (1, 0)),
|
|
("w", (-1, 0)),
|
|
("west", (-1, 0)),
|
|
("ne", (1, -1)),
|
|
("northeast", (1, -1)),
|
|
("nw", (-1, -1)),
|
|
("northwest", (-1, -1)),
|
|
("se", (1, 1)),
|
|
("southeast", (1, 1)),
|
|
("sw", (-1, 1)),
|
|
("southwest", (-1, 1)),
|
|
],
|
|
)
|
|
def test_direction_deltas(direction, expected_delta):
|
|
"""Test that all direction commands map to correct deltas."""
|
|
assert movement.DIRECTIONS[direction] == expected_delta
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_movement_updates_position(player, test_zone):
|
|
"""Test that movement updates player position when passable."""
|
|
# Clear players registry to avoid test pollution
|
|
from mudlib.player import players
|
|
|
|
players.clear()
|
|
|
|
original_x, original_y = player.x, player.y
|
|
await movement.move_north(player, "")
|
|
|
|
assert player.x == original_x
|
|
assert player.y == original_y - 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_movement_blocked_by_impassable_terrain(player, test_zone, mock_writer):
|
|
"""Test that movement is blocked by impassable terrain."""
|
|
# Make the target position impassable
|
|
target_y = player.y - 1
|
|
test_zone.terrain[target_y][player.x] = "^" # mountain
|
|
test_zone.impassable = {"^"}
|
|
|
|
original_x, original_y = player.x, player.y
|
|
await movement.move_north(player, "")
|
|
|
|
# Position should not change
|
|
assert player.x == original_x
|
|
assert player.y == original_y
|
|
|
|
# Should send a message to the player
|
|
assert mock_writer.write.called
|
|
error_msg = mock_writer.write.call_args[0][0]
|
|
assert "can't" in error_msg.lower() or "cannot" in error_msg.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_movement_sends_departure_message(player, test_zone):
|
|
"""Test that movement sends departure message to nearby players."""
|
|
# Create another player in the area
|
|
other_writer = MagicMock()
|
|
other_writer.write = MagicMock()
|
|
other_writer.drain = AsyncMock()
|
|
other_player = Player(
|
|
name="OtherPlayer", x=5, y=4, reader=MagicMock(), writer=other_writer
|
|
)
|
|
other_player.location = test_zone
|
|
test_zone._contents.append(other_player)
|
|
|
|
# Register both players
|
|
from mudlib.player import players
|
|
|
|
players.clear()
|
|
players[player.name] = player
|
|
players[other_player.name] = other_player
|
|
|
|
await movement.move_north(player, "")
|
|
|
|
# Other player should have received a departure message
|
|
# (We'll check this is called, exact message format is implementation detail)
|
|
assert other_player.writer.write.called
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_arrival_message_uses_opposite_direction(player, test_zone):
|
|
"""Test that arrival messages use the opposite direction."""
|
|
# Create another player at the destination
|
|
other_writer = MagicMock()
|
|
other_writer.write = MagicMock()
|
|
other_writer.drain = AsyncMock()
|
|
other_player = Player(
|
|
name="OtherPlayer", x=5, y=3, reader=MagicMock(), writer=other_writer
|
|
)
|
|
other_player.location = test_zone
|
|
test_zone._contents.append(other_player)
|
|
|
|
from mudlib.player import players
|
|
|
|
players.clear()
|
|
players[player.name] = player
|
|
players[other_player.name] = other_player
|
|
|
|
# Player at (5, 5) moves north to (5, 4)
|
|
await movement.move_north(player, "")
|
|
|
|
# Other player at (5, 3) should see arrival "from the south"
|
|
# (Implementation will determine exact message format)
|
|
assert other_player.writer.write.called
|
|
|
|
|
|
# Test look command
|
|
@pytest.mark.asyncio
|
|
async def test_look_command_sends_viewport(player, test_zone):
|
|
"""Test that look command sends the viewport to the player."""
|
|
await look.cmd_look(player, "")
|
|
|
|
assert player.writer.write.called
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_look_command_shows_player_at_center(player, test_zone):
|
|
"""Test that look command shows player @ at center."""
|
|
await look.cmd_look(player, "")
|
|
|
|
# Check that the output contains the @ symbol for the player
|
|
output = "".join([call[0][0] for call in player.writer.write.call_args_list])
|
|
assert "@" in output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_look_command_shows_other_players(player, test_zone):
|
|
"""Test that look command shows other players as *."""
|
|
# Create another player in the viewport
|
|
other_player = Player(
|
|
name="OtherPlayer",
|
|
x=6,
|
|
y=5,
|
|
reader=MagicMock(),
|
|
writer=MagicMock(),
|
|
)
|
|
other_player.location = test_zone
|
|
test_zone._contents.append(other_player)
|
|
|
|
from mudlib.player import players
|
|
|
|
players.clear()
|
|
players[player.name] = player
|
|
players[other_player.name] = other_player
|
|
|
|
await look.cmd_look(player, "")
|
|
|
|
# Check that the output contains * for other players
|
|
output = "".join([call[0][0] for call in player.writer.write.call_args_list])
|
|
assert "*" in output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_look_shows_effects_on_viewport(player, test_zone):
|
|
"""Test that active effects overlay on the viewport."""
|
|
from mudlib.player import players
|
|
|
|
players.clear()
|
|
players[player.name] = player
|
|
|
|
active_effects.clear()
|
|
# place a cloud effect 2 tiles east of the player (viewport center is 10,5)
|
|
# player is at (5, 5), so effect at (7, 5) -> viewport (12, 5)
|
|
add_effect(7, 5, "~", "\033[1;97m", ttl=10.0)
|
|
|
|
await look.cmd_look(player, "")
|
|
|
|
output = "".join([call[0][0] for call in player.writer.write.call_args_list])
|
|
# the output should contain our cloud color applied to ~
|
|
# (bright white ~ instead of blue ~)
|
|
assert "\033[1;97m~" + RESET in output
|
|
|
|
active_effects.clear()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_effects_dont_override_player_marker(player, test_zone):
|
|
"""Effects at the player's position should not hide the @ marker."""
|
|
from mudlib.player import players
|
|
|
|
players.clear()
|
|
players[player.name] = player
|
|
|
|
active_effects.clear()
|
|
# place an effect right at the player's position
|
|
add_effect(player.x, player.y, "~", "\033[1;97m", ttl=10.0)
|
|
|
|
await look.cmd_look(player, "")
|
|
|
|
output = "".join([call[0][0] for call in player.writer.write.call_args_list])
|
|
# player @ should still be visible at center
|
|
assert "@" in output
|
|
|
|
active_effects.clear()
|
|
|
|
|
|
# Test mode stack
|
|
def test_mode_stack_default(player):
|
|
"""Player starts in normal mode."""
|
|
assert player.mode == "normal"
|
|
assert player.mode_stack == ["normal"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatch_blocks_wrong_mode(player, mock_writer):
|
|
"""Commands with wrong mode get rejected."""
|
|
|
|
async def combat_handler(p, args):
|
|
pass
|
|
|
|
commands.register(CommandDefinition("strike", combat_handler, mode="combat"))
|
|
await commands.dispatch(player, "strike")
|
|
|
|
assert mock_writer.write.called
|
|
written = mock_writer.write.call_args[0][0]
|
|
assert "can't" in written.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatch_allows_wildcard_mode(player):
|
|
"""Commands with mode='*' work from any mode."""
|
|
called = False
|
|
|
|
async def any_handler(p, args):
|
|
nonlocal called
|
|
called = True
|
|
|
|
commands.register(CommandDefinition("universal", any_handler, mode="*"))
|
|
await commands.dispatch(player, "universal")
|
|
|
|
assert called
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatch_allows_matching_mode(player):
|
|
"""Commands work when player mode matches command mode."""
|
|
called = False
|
|
|
|
async def combat_handler(p, args):
|
|
nonlocal called
|
|
called = True
|
|
|
|
commands.register(CommandDefinition("strike", combat_handler, mode="combat"))
|
|
player.mode_stack.append("combat")
|
|
await commands.dispatch(player, "strike")
|
|
|
|
assert called
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_look_shows_ground_items(player, test_zone):
|
|
"""look shows things on the ground at the player's position."""
|
|
from mudlib.player import players
|
|
from mudlib.thing import Thing
|
|
|
|
players.clear()
|
|
players[player.name] = player
|
|
|
|
Thing(name="rock", location=test_zone, x=5, y=5)
|
|
await look.cmd_look(player, "")
|
|
|
|
output = "".join([call[0][0] for call in player.writer.write.call_args_list])
|
|
assert "rock" in output.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_look_shows_multiple_ground_items(player, test_zone):
|
|
"""look shows all things at the player's position."""
|
|
from mudlib.player import players
|
|
from mudlib.thing import Thing
|
|
|
|
players.clear()
|
|
players[player.name] = player
|
|
|
|
Thing(name="rock", location=test_zone, x=5, y=5)
|
|
Thing(name="sword", location=test_zone, x=5, y=5)
|
|
await look.cmd_look(player, "")
|
|
|
|
output = "".join([call[0][0] for call in player.writer.write.call_args_list])
|
|
assert "rock" in output.lower()
|
|
assert "sword" in output.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_look_no_ground_items_no_extra_output(player, test_zone):
|
|
"""look with no ground items doesn't mention items."""
|
|
from mudlib.player import players
|
|
|
|
players.clear()
|
|
players[player.name] = player
|
|
|
|
await look.cmd_look(player, "")
|
|
|
|
output = "".join([call[0][0] for call in player.writer.write.call_args_list])
|
|
assert "on the ground" not in output.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_look_ignores_items_at_other_positions(player, test_zone):
|
|
"""look doesn't show items that are at different positions."""
|
|
from mudlib.player import players
|
|
from mudlib.thing import Thing
|
|
|
|
players.clear()
|
|
players[player.name] = player
|
|
|
|
Thing(name="far_rock", location=test_zone, x=9, y=9)
|
|
await look.cmd_look(player, "")
|
|
|
|
output = "".join([call[0][0] for call in player.writer.write.call_args_list])
|
|
assert "far_rock" not in output.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_quit_removes_player_from_zone(monkeypatch):
|
|
"""Quitting removes player from zone contents."""
|
|
from mudlib.commands import quit as quit_mod
|
|
from mudlib.player import players
|
|
|
|
monkeypatch.setattr(quit_mod, "save_player", lambda p: None)
|
|
|
|
terrain = [["." for _ in range(10)] for _ in range(10)]
|
|
zone = Zone(
|
|
name="qz",
|
|
width=10,
|
|
height=10,
|
|
toroidal=True,
|
|
terrain=terrain,
|
|
impassable=set(),
|
|
)
|
|
writer = MagicMock()
|
|
writer.write = MagicMock()
|
|
writer.drain = AsyncMock()
|
|
writer.close = MagicMock()
|
|
p = Player(
|
|
name="quitter",
|
|
location=zone,
|
|
x=5,
|
|
y=5,
|
|
reader=MagicMock(),
|
|
writer=writer,
|
|
)
|
|
players.clear()
|
|
players["quitter"] = p
|
|
|
|
assert p in zone._contents
|
|
await quit_mod.cmd_quit(p, "")
|
|
assert p not in zone._contents
|
|
assert "quitter" not in players
|