Compare commits
8 commits
058ba1b7de
...
f418805b78
| Author | SHA1 | Date | |
|---|---|---|---|
| f418805b78 | |||
| 64c25b1025 | |||
| 3d386fbf99 | |||
| ca282851e3 | |||
| 6ed71873b5 | |||
| e247d70612 | |||
| e9f70ebd2f | |||
| d253012122 |
15 changed files with 708 additions and 13 deletions
|
|
@ -80,8 +80,14 @@ async def do_attack(player: Player, target_args: str, move: CombatMove) -> None:
|
|||
CombatState.WINDOW,
|
||||
)
|
||||
|
||||
# Execute the attack
|
||||
# Execute the attack (deducts stamina)
|
||||
encounter.attack(move)
|
||||
|
||||
# Send vitals update immediately after stamina deduction
|
||||
from mudlib.gmcp import send_char_vitals
|
||||
|
||||
send_char_vitals(player)
|
||||
|
||||
if switching:
|
||||
await player.send(f"You switch to {move.name}!\r\n")
|
||||
else:
|
||||
|
|
@ -106,6 +112,11 @@ async def do_defend(player: Player, _args: str, move: CombatMove) -> None:
|
|||
|
||||
player.stamina -= move.stamina_cost
|
||||
|
||||
# Send vitals update immediately after stamina deduction
|
||||
from mudlib.gmcp import send_char_vitals
|
||||
|
||||
send_char_vitals(player)
|
||||
|
||||
# If in combat, queue the defense on the encounter
|
||||
encounter = get_encounter(player)
|
||||
if encounter is not None:
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import time
|
|||
|
||||
from mudlib.combat.encounter import IDLE_TIMEOUT, CombatEncounter, CombatState
|
||||
from mudlib.entity import Entity, Mob
|
||||
from mudlib.gmcp import send_char_status, send_char_vitals
|
||||
|
||||
# Global list of active combat encounters
|
||||
active_encounters: list[CombatEncounter] = []
|
||||
|
|
@ -85,6 +86,7 @@ async def process_combat() -> None:
|
|||
for entity in (encounter.attacker, encounter.defender):
|
||||
if isinstance(entity, Player) and entity.mode == "combat":
|
||||
entity.mode_stack.pop()
|
||||
send_char_status(entity)
|
||||
|
||||
end_encounter(encounter)
|
||||
continue
|
||||
|
|
@ -100,6 +102,13 @@ async def process_combat() -> None:
|
|||
await encounter.attacker.send(result.attacker_msg + "\r\n")
|
||||
await encounter.defender.send(result.defender_msg + "\r\n")
|
||||
|
||||
# Send vitals update after damage resolution
|
||||
from mudlib.player import Player
|
||||
|
||||
for entity in (encounter.attacker, encounter.defender):
|
||||
if isinstance(entity, Player):
|
||||
send_char_vitals(entity)
|
||||
|
||||
if result.combat_ended:
|
||||
# Determine winner/loser
|
||||
if encounter.defender.pl <= 0:
|
||||
|
|
@ -126,10 +135,12 @@ async def process_combat() -> None:
|
|||
attacker = encounter.attacker
|
||||
if isinstance(attacker, Player) and attacker.mode == "combat":
|
||||
attacker.mode_stack.pop()
|
||||
send_char_status(attacker)
|
||||
|
||||
defender = encounter.defender
|
||||
if isinstance(defender, Player) and defender.mode == "combat":
|
||||
defender.mode_stack.pop()
|
||||
send_char_status(defender)
|
||||
|
||||
# Remove encounter from active list
|
||||
end_encounter(encounter)
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
|
||||
from mudlib.commands import CommandDefinition, register
|
||||
from mudlib.entity import Entity
|
||||
from mudlib.gmcp import send_map_data, send_room_info
|
||||
from mudlib.player import Player
|
||||
from mudlib.portal import Portal
|
||||
from mudlib.zone import Zone
|
||||
|
|
@ -106,6 +107,8 @@ async def move_player(player: Player, dx: int, dy: int, direction_name: str) ->
|
|||
from mudlib.commands.look import cmd_look
|
||||
|
||||
await cmd_look(player, "")
|
||||
send_room_info(player)
|
||||
send_map_data(player)
|
||||
return # Don't do normal arrival+look
|
||||
else:
|
||||
await player.send("The portal doesn't lead anywhere.\r\n")
|
||||
|
|
@ -124,6 +127,8 @@ async def move_player(player: Player, dx: int, dy: int, direction_name: str) ->
|
|||
from mudlib.commands.look import cmd_look
|
||||
|
||||
await cmd_look(player, "")
|
||||
send_room_info(player)
|
||||
send_map_data(player)
|
||||
|
||||
|
||||
async def send_nearby_message(entity: Entity, x: int, y: int, message: str) -> None:
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
"""Rest command for restoring stamina."""
|
||||
|
||||
from mudlib.commands.movement import send_nearby_message
|
||||
from mudlib.gmcp import send_char_status
|
||||
from mudlib.player import Player
|
||||
|
||||
|
||||
|
|
@ -20,6 +21,7 @@ async def cmd_rest(player: Player, args: str) -> None:
|
|||
if player.resting:
|
||||
# Stop resting
|
||||
player.resting = False
|
||||
send_char_status(player)
|
||||
await player.send("You stop resting.\r\n")
|
||||
await send_nearby_message(
|
||||
player, player.x, player.y, f"{player.name} stops resting.\r\n"
|
||||
|
|
@ -27,6 +29,7 @@ async def cmd_rest(player: Player, args: str) -> None:
|
|||
else:
|
||||
# Start resting
|
||||
player.resting = True
|
||||
send_char_status(player)
|
||||
await player.send("You begin to rest.\r\n")
|
||||
await send_nearby_message(
|
||||
player, player.x, player.y, f"{player.name} begins to rest.\r\n"
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ class Entity(Object):
|
|||
y: int = 0
|
||||
# Combat stats
|
||||
pl: float = 100.0 # power level (health and damage multiplier)
|
||||
max_pl: float = 100.0 # maximum power level
|
||||
stamina: float = 100.0 # current stamina
|
||||
max_stamina: float = 100.0 # stamina ceiling
|
||||
defense_locked_until: float = 0.0 # monotonic time when defense recovery ends
|
||||
|
|
|
|||
118
src/mudlib/gmcp.py
Normal file
118
src/mudlib/gmcp.py
Normal file
|
|
@ -0,0 +1,118 @@
|
|||
"""GMCP and MSDP package support for rich MUD clients."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from mudlib.player import Player
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def send_char_vitals(player: Player) -> None:
|
||||
"""Send Char.Vitals — pl, max_pl, stamina, max_stamina."""
|
||||
player.send_gmcp(
|
||||
"Char.Vitals",
|
||||
{
|
||||
"pl": round(player.pl, 1),
|
||||
"max_pl": round(player.max_pl, 1),
|
||||
"stamina": round(player.stamina, 1),
|
||||
"max_stamina": round(player.max_stamina, 1),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def send_char_status(player: Player) -> None:
|
||||
"""Send Char.Status — flying, resting, mode, in_combat."""
|
||||
player.send_gmcp(
|
||||
"Char.Status",
|
||||
{
|
||||
"flying": player.flying,
|
||||
"resting": player.resting,
|
||||
"mode": player.mode,
|
||||
"in_combat": player.mode == "combat",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def send_room_info(player: Player) -> None:
|
||||
"""Send Room.Info — zone, coordinates, terrain, exits."""
|
||||
from mudlib.zone import Zone
|
||||
|
||||
zone = player.location
|
||||
if not isinstance(zone, Zone):
|
||||
return
|
||||
|
||||
terrain = zone.get_tile(player.x, player.y)
|
||||
|
||||
# Build exits list from passable adjacent tiles
|
||||
exits = []
|
||||
directions = [
|
||||
("north", 0, -1),
|
||||
("south", 0, 1),
|
||||
("east", 1, 0),
|
||||
("west", -1, 0),
|
||||
("northeast", 1, -1),
|
||||
("northwest", -1, -1),
|
||||
("southeast", 1, 1),
|
||||
("southwest", -1, 1),
|
||||
]
|
||||
for name, dx, dy in directions:
|
||||
nx, ny = player.x + dx, player.y + dy
|
||||
if zone.is_passable(nx, ny):
|
||||
exits.append(name)
|
||||
|
||||
player.send_gmcp(
|
||||
"Room.Info",
|
||||
{
|
||||
"zone": zone.name,
|
||||
"x": player.x,
|
||||
"y": player.y,
|
||||
"terrain": terrain,
|
||||
"exits": exits,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def send_map_data(player: Player) -> None:
|
||||
"""Send Room.Map — terrain viewport around player for client rendering."""
|
||||
from mudlib.zone import Zone
|
||||
|
||||
zone = player.location
|
||||
if not isinstance(zone, Zone):
|
||||
return
|
||||
|
||||
radius = 10
|
||||
rows = []
|
||||
for dy in range(-radius, radius + 1):
|
||||
row = []
|
||||
for dx in range(-radius, radius + 1):
|
||||
tx = player.x + dx
|
||||
ty = player.y + dy
|
||||
wx, wy = zone.wrap(tx, ty)
|
||||
row.append(zone.terrain[wy][wx])
|
||||
rows.append(row)
|
||||
|
||||
player.send_gmcp(
|
||||
"Room.Map",
|
||||
{
|
||||
"x": player.x,
|
||||
"y": player.y,
|
||||
"radius": radius,
|
||||
"terrain": rows,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def send_msdp_vitals(player: Player) -> None:
|
||||
"""Send MSDP variable updates for real-time gauges."""
|
||||
player.send_msdp(
|
||||
{
|
||||
"HEALTH": str(round(player.pl, 1)),
|
||||
"HEALTH_MAX": str(round(player.max_pl, 1)),
|
||||
"STAMINA": str(round(player.stamina, 1)),
|
||||
"STAMINA_MAX": str(round(player.max_stamina, 1)),
|
||||
}
|
||||
)
|
||||
|
|
@ -44,6 +44,16 @@ class Player(Entity):
|
|||
self.writer.write(message)
|
||||
await self.writer.drain()
|
||||
|
||||
def send_gmcp(self, package: str, data: Any = None) -> None:
|
||||
"""Send a GMCP message to the client (no-op if unsupported)."""
|
||||
if self.writer is not None:
|
||||
self.writer.send_gmcp(package, data)
|
||||
|
||||
def send_msdp(self, variables: dict) -> None:
|
||||
"""Send MSDP variables to the client (no-op if unsupported)."""
|
||||
if self.writer is not None:
|
||||
self.writer.send_msdp(variables)
|
||||
|
||||
|
||||
# Global registry of connected players
|
||||
players: dict[str, Player] = {}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
"""Resting system for stamina regeneration."""
|
||||
|
||||
from mudlib.commands.movement import send_nearby_message
|
||||
from mudlib.gmcp import send_char_status, send_char_vitals
|
||||
from mudlib.player import players
|
||||
|
||||
# Stamina regeneration rate: 2.0 per second
|
||||
|
|
@ -25,6 +26,8 @@ async def process_resting() -> None:
|
|||
# Check if we reached max stamina
|
||||
if player.stamina >= player.max_stamina:
|
||||
player.resting = False
|
||||
send_char_status(player)
|
||||
send_char_vitals(player)
|
||||
await player.send("You feel fully rested.\r\n")
|
||||
await send_nearby_message(
|
||||
player, player.x, player.y, f"{player.name} stops resting.\r\n"
|
||||
|
|
|
|||
|
|
@ -32,6 +32,13 @@ from mudlib.combat.commands import register_combat_commands
|
|||
from mudlib.combat.engine import process_combat
|
||||
from mudlib.content import load_commands
|
||||
from mudlib.effects import clear_expired
|
||||
from mudlib.gmcp import (
|
||||
send_char_status,
|
||||
send_char_vitals,
|
||||
send_map_data,
|
||||
send_msdp_vitals,
|
||||
send_room_info,
|
||||
)
|
||||
from mudlib.if_session import broadcast_to_spectators
|
||||
from mudlib.mob_ai import process_mobs
|
||||
from mudlib.mobs import load_mob_templates, mob_templates
|
||||
|
|
@ -77,14 +84,21 @@ async def game_loop() -> None:
|
|||
"""Run periodic game tasks at TICK_RATE ticks per second."""
|
||||
log.info("game loop started (%d ticks/sec)", TICK_RATE)
|
||||
last_save_time = time.monotonic()
|
||||
tick_count = 0
|
||||
|
||||
while True:
|
||||
t0 = asyncio.get_event_loop().time()
|
||||
tick_count += 1
|
||||
clear_expired()
|
||||
await process_combat()
|
||||
await process_mobs(mudlib.combat.commands.combat_moves)
|
||||
await process_resting()
|
||||
|
||||
# MSDP updates once per second (every TICK_RATE ticks)
|
||||
if tick_count % TICK_RATE == 0:
|
||||
for p in list(players.values()):
|
||||
send_msdp_vitals(p)
|
||||
|
||||
# Periodic auto-save (every 60 seconds)
|
||||
current_time = time.monotonic()
|
||||
if current_time - last_save_time >= AUTOSAVE_INTERVAL:
|
||||
|
|
@ -347,6 +361,12 @@ async def shell(
|
|||
# Show initial map
|
||||
await mudlib.commands.look.cmd_look(player, "")
|
||||
|
||||
# Send initial GMCP data to rich clients
|
||||
send_char_vitals(player)
|
||||
send_char_status(player)
|
||||
send_room_info(player)
|
||||
send_map_data(player)
|
||||
|
||||
# Command loop
|
||||
try:
|
||||
while not _writer.is_closing():
|
||||
|
|
|
|||
504
tests/test_gmcp.py
Normal file
504
tests/test_gmcp.py
Normal file
|
|
@ -0,0 +1,504 @@
|
|||
"""Tests for GMCP and MSDP support."""
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from mudlib.gmcp import (
|
||||
send_char_status,
|
||||
send_char_vitals,
|
||||
send_map_data,
|
||||
send_msdp_vitals,
|
||||
send_room_info,
|
||||
)
|
||||
from mudlib.player import Player, players
|
||||
from mudlib.zone import Zone
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def clear_state():
|
||||
"""Clear players before and after each test."""
|
||||
players.clear()
|
||||
yield
|
||||
players.clear()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_zone():
|
||||
"""Create a small test zone with known terrain."""
|
||||
terrain = [["." for _ in range(25)] for _ in range(25)] # 25x25 simple terrain
|
||||
# Add some impassable tiles for exit testing
|
||||
terrain[5][10] = "^" # mountain north of player
|
||||
zone = Zone(
|
||||
name="testzone",
|
||||
width=25,
|
||||
height=25,
|
||||
toroidal=True,
|
||||
terrain=terrain,
|
||||
impassable={"^", "~"},
|
||||
)
|
||||
return zone
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_writer():
|
||||
writer = MagicMock()
|
||||
writer.write = MagicMock()
|
||||
writer.drain = AsyncMock()
|
||||
writer.send_gmcp = MagicMock()
|
||||
writer.send_msdp = MagicMock()
|
||||
return writer
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def player(mock_writer, test_zone):
|
||||
p = Player(name="Goku", x=10, y=10, writer=mock_writer)
|
||||
p.location = test_zone
|
||||
p.pl = 85.5
|
||||
p.max_pl = 100.0
|
||||
p.stamina = 42.3
|
||||
p.max_stamina = 100.0
|
||||
players[p.name] = p
|
||||
return p
|
||||
|
||||
|
||||
def test_send_char_vitals(player):
|
||||
"""Test Char.Vitals sends correct payload."""
|
||||
send_char_vitals(player)
|
||||
|
||||
player.writer.send_gmcp.assert_called_once_with(
|
||||
"Char.Vitals",
|
||||
{
|
||||
"pl": 85.5,
|
||||
"max_pl": 100.0,
|
||||
"stamina": 42.3,
|
||||
"max_stamina": 100.0,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def test_send_char_status_normal_mode(player):
|
||||
"""Test Char.Status sends correct payload in normal mode."""
|
||||
player.flying = False
|
||||
player.resting = False
|
||||
player.mode_stack = ["normal"]
|
||||
|
||||
send_char_status(player)
|
||||
|
||||
player.writer.send_gmcp.assert_called_once_with(
|
||||
"Char.Status",
|
||||
{
|
||||
"flying": False,
|
||||
"resting": False,
|
||||
"mode": "normal",
|
||||
"in_combat": False,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def test_send_char_status_in_combat(player):
|
||||
"""Test Char.Status detects combat mode."""
|
||||
player.mode_stack = ["normal", "combat"]
|
||||
|
||||
send_char_status(player)
|
||||
|
||||
args = player.writer.send_gmcp.call_args[0]
|
||||
assert args[0] == "Char.Status"
|
||||
assert args[1]["mode"] == "combat"
|
||||
assert args[1]["in_combat"] is True
|
||||
|
||||
|
||||
def test_send_char_status_flying_and_resting(player):
|
||||
"""Test Char.Status reflects flying and resting states."""
|
||||
player.flying = True
|
||||
player.resting = True
|
||||
|
||||
send_char_status(player)
|
||||
|
||||
args = player.writer.send_gmcp.call_args[0]
|
||||
assert args[1]["flying"] is True
|
||||
assert args[1]["resting"] is True
|
||||
|
||||
|
||||
def test_send_room_info(player):
|
||||
"""Test Room.Info sends zone, coordinates, terrain, and exits."""
|
||||
send_room_info(player)
|
||||
|
||||
player.writer.send_gmcp.assert_called_once()
|
||||
args = player.writer.send_gmcp.call_args[0]
|
||||
assert args[0] == "Room.Info"
|
||||
|
||||
data = args[1]
|
||||
assert data["zone"] == "testzone"
|
||||
assert data["x"] == 10
|
||||
assert data["y"] == 10
|
||||
assert data["terrain"] == "."
|
||||
# All adjacent tiles should be passable (player at 10,10, all tiles are ".")
|
||||
assert "north" in data["exits"]
|
||||
assert "south" in data["exits"]
|
||||
assert "east" in data["exits"]
|
||||
assert "west" in data["exits"]
|
||||
assert "northeast" in data["exits"]
|
||||
assert "northwest" in data["exits"]
|
||||
assert "southeast" in data["exits"]
|
||||
assert "southwest" in data["exits"]
|
||||
|
||||
|
||||
def test_send_room_info_blocked_exit(player, test_zone):
|
||||
"""Test Room.Info excludes blocked exits."""
|
||||
# Put player next to mountain at 10,5
|
||||
player.y = 6
|
||||
test_zone.terrain[5][10] = "^" # mountain to the north
|
||||
|
||||
send_room_info(player)
|
||||
|
||||
args = player.writer.send_gmcp.call_args[0]
|
||||
data = args[1]
|
||||
# North (y-1=5) should be impassable
|
||||
assert "north" not in data["exits"]
|
||||
assert "south" in data["exits"]
|
||||
|
||||
|
||||
def test_send_room_info_no_zone(player):
|
||||
"""Test Room.Info with no zone does not crash."""
|
||||
player.location = None
|
||||
|
||||
send_room_info(player)
|
||||
|
||||
# Should not have called send_gmcp
|
||||
player.writer.send_gmcp.assert_not_called()
|
||||
|
||||
|
||||
def test_send_map_data(player):
|
||||
"""Test Room.Map sends terrain grid."""
|
||||
send_map_data(player)
|
||||
|
||||
player.writer.send_gmcp.assert_called_once()
|
||||
args = player.writer.send_gmcp.call_args[0]
|
||||
assert args[0] == "Room.Map"
|
||||
|
||||
data = args[1]
|
||||
assert data["x"] == 10
|
||||
assert data["y"] == 10
|
||||
assert data["radius"] == 10
|
||||
# Grid should be 21x21 (radius 10 = -10 to +10 inclusive)
|
||||
assert len(data["terrain"]) == 21
|
||||
assert len(data["terrain"][0]) == 21
|
||||
# Center tile should be "."
|
||||
assert data["terrain"][10][10] == "."
|
||||
|
||||
|
||||
def test_send_map_data_toroidal_wrapping(player, test_zone):
|
||||
"""Test Room.Map wraps correctly in toroidal zones."""
|
||||
# Put player near edge to test wrapping
|
||||
player.x = 2
|
||||
player.y = 2
|
||||
|
||||
send_map_data(player)
|
||||
|
||||
args = player.writer.send_gmcp.call_args[0]
|
||||
data = args[1]
|
||||
# Should still be 21x21 even near edge (wraps around)
|
||||
assert len(data["terrain"]) == 21
|
||||
assert len(data["terrain"][0]) == 21
|
||||
|
||||
|
||||
def test_send_map_data_no_zone(player):
|
||||
"""Test Room.Map with no zone does not crash."""
|
||||
player.location = None
|
||||
|
||||
send_map_data(player)
|
||||
|
||||
player.writer.send_gmcp.assert_not_called()
|
||||
|
||||
|
||||
def test_send_msdp_vitals(player):
|
||||
"""Test MSDP sends vitals with correct variable names."""
|
||||
send_msdp_vitals(player)
|
||||
|
||||
player.writer.send_msdp.assert_called_once_with(
|
||||
{
|
||||
"HEALTH": "85.5",
|
||||
"HEALTH_MAX": "100.0",
|
||||
"STAMINA": "42.3",
|
||||
"STAMINA_MAX": "100.0",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_player_send_gmcp(player):
|
||||
"""Test Player.send_gmcp convenience method."""
|
||||
player.send_gmcp("Test.Package", {"key": "value"})
|
||||
|
||||
player.writer.send_gmcp.assert_called_once_with("Test.Package", {"key": "value"})
|
||||
|
||||
|
||||
def test_player_send_gmcp_no_writer():
|
||||
"""Test Player.send_gmcp with no writer does not crash."""
|
||||
p = Player(name="NoWriter", writer=None)
|
||||
p.send_gmcp("Test.Package", {}) # Should not raise
|
||||
|
||||
|
||||
def test_player_send_msdp(player):
|
||||
"""Test Player.send_msdp convenience method."""
|
||||
player.send_msdp({"VAR": "value"})
|
||||
|
||||
player.writer.send_msdp.assert_called_once_with({"VAR": "value"})
|
||||
|
||||
|
||||
def test_player_send_msdp_no_writer():
|
||||
"""Test Player.send_msdp with no writer does not crash."""
|
||||
p = Player(name="NoWriter", writer=None)
|
||||
p.send_msdp({}) # Should not raise
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_room_info_sent_on_movement(player, test_zone):
|
||||
"""Test Room.Info is sent when player moves."""
|
||||
from mudlib.commands.movement import move_player
|
||||
|
||||
await move_player(player, 1, 0, "east")
|
||||
|
||||
# Verify send_gmcp was called with Room.Info
|
||||
gmcp_calls = player.writer.send_gmcp.call_args_list
|
||||
room_info_calls = [call for call in gmcp_calls if call[0][0] == "Room.Info"]
|
||||
assert len(room_info_calls) > 0, "Room.Info should be sent on movement"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_map_data_sent_on_movement(player, test_zone):
|
||||
"""Test Room.Map is sent when player moves."""
|
||||
from mudlib.commands.movement import move_player
|
||||
|
||||
await move_player(player, 1, 0, "east")
|
||||
|
||||
# Verify send_gmcp was called with Room.Map
|
||||
gmcp_calls = player.writer.send_gmcp.call_args_list
|
||||
room_map_calls = [call for call in gmcp_calls if call[0][0] == "Room.Map"]
|
||||
assert len(room_map_calls) > 0, "Room.Map should be sent on movement"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_char_vitals_sent_on_rest_complete(player):
|
||||
"""Test Char.Vitals is sent when resting completes."""
|
||||
from mudlib.resting import process_resting
|
||||
|
||||
player.resting = True
|
||||
player.stamina = player.max_stamina - 0.1 # Almost full
|
||||
|
||||
await process_resting()
|
||||
|
||||
# Should have called send_gmcp with both Char.Status and Char.Vitals
|
||||
assert player.writer.send_gmcp.call_count == 2
|
||||
|
||||
# First call should be Char.Status
|
||||
first_call = player.writer.send_gmcp.call_args_list[0]
|
||||
assert first_call[0][0] == "Char.Status"
|
||||
|
||||
# Second call should be Char.Vitals
|
||||
second_call = player.writer.send_gmcp.call_args_list[1]
|
||||
assert second_call[0][0] == "Char.Vitals"
|
||||
assert second_call[0][1] == {
|
||||
"pl": round(player.pl, 1),
|
||||
"max_pl": round(player.max_pl, 1),
|
||||
"stamina": round(player.max_stamina, 1),
|
||||
"max_stamina": round(player.max_stamina, 1),
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_char_vitals_sent_on_combat_resolve():
|
||||
"""Test Char.Vitals is sent to both players on combat resolution."""
|
||||
import time
|
||||
|
||||
from mudlib.combat.engine import active_encounters, process_combat, start_encounter
|
||||
from mudlib.combat.moves import CombatMove
|
||||
|
||||
# Clear encounters
|
||||
active_encounters.clear()
|
||||
|
||||
# Create two players with mock writers
|
||||
mock_writer_1 = MagicMock()
|
||||
mock_writer_1.write = MagicMock()
|
||||
mock_writer_1.drain = AsyncMock()
|
||||
mock_writer_1.send_gmcp = MagicMock()
|
||||
|
||||
mock_writer_2 = MagicMock()
|
||||
mock_writer_2.write = MagicMock()
|
||||
mock_writer_2.drain = AsyncMock()
|
||||
mock_writer_2.send_gmcp = MagicMock()
|
||||
|
||||
attacker = Player(
|
||||
name="Goku", x=0, y=0, pl=100.0, stamina=50.0, writer=mock_writer_1
|
||||
)
|
||||
defender = Player(
|
||||
name="Vegeta", x=0, y=0, pl=100.0, stamina=50.0, writer=mock_writer_2
|
||||
)
|
||||
|
||||
attacker.mode_stack.append("combat")
|
||||
defender.mode_stack.append("combat")
|
||||
|
||||
# Create encounter and attack
|
||||
encounter = start_encounter(attacker, defender)
|
||||
punch = CombatMove(
|
||||
name="punch right",
|
||||
move_type="attack",
|
||||
stamina_cost=5.0,
|
||||
timing_window_ms=800,
|
||||
damage_pct=0.15,
|
||||
countered_by=["dodge left"],
|
||||
)
|
||||
encounter.attack(punch)
|
||||
|
||||
# Advance past telegraph and window to trigger resolution
|
||||
time.sleep(0.31)
|
||||
await process_combat()
|
||||
time.sleep(0.85)
|
||||
|
||||
# Reset mocks before the resolution call
|
||||
mock_writer_1.send_gmcp.reset_mock()
|
||||
mock_writer_2.send_gmcp.reset_mock()
|
||||
|
||||
await process_combat()
|
||||
|
||||
# Both players should have received Char.Vitals
|
||||
assert mock_writer_1.send_gmcp.call_count == 1
|
||||
assert mock_writer_2.send_gmcp.call_count == 1
|
||||
|
||||
# Check attacker's call
|
||||
attacker_call = mock_writer_1.send_gmcp.call_args[0]
|
||||
assert attacker_call[0] == "Char.Vitals"
|
||||
assert "pl" in attacker_call[1]
|
||||
assert "max_pl" in attacker_call[1]
|
||||
assert "stamina" in attacker_call[1]
|
||||
assert "max_stamina" in attacker_call[1]
|
||||
|
||||
# Check defender's call
|
||||
defender_call = mock_writer_2.send_gmcp.call_args[0]
|
||||
assert defender_call[0] == "Char.Vitals"
|
||||
assert "pl" in defender_call[1]
|
||||
assert "max_pl" in defender_call[1]
|
||||
assert "stamina" in defender_call[1]
|
||||
assert "max_stamina" in defender_call[1]
|
||||
|
||||
# Cleanup
|
||||
active_encounters.clear()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_char_status_sent_on_combat_end():
|
||||
"""Test Char.Status is sent when combat ends (victory/defeat)."""
|
||||
import time
|
||||
|
||||
from mudlib.combat.engine import active_encounters, process_combat, start_encounter
|
||||
from mudlib.combat.moves import CombatMove
|
||||
|
||||
# Clear encounters
|
||||
active_encounters.clear()
|
||||
|
||||
# Create two players with mock writers
|
||||
mock_writer_1 = MagicMock()
|
||||
mock_writer_1.write = MagicMock()
|
||||
mock_writer_1.drain = AsyncMock()
|
||||
mock_writer_1.send_gmcp = MagicMock()
|
||||
|
||||
mock_writer_2 = MagicMock()
|
||||
mock_writer_2.write = MagicMock()
|
||||
mock_writer_2.drain = AsyncMock()
|
||||
mock_writer_2.send_gmcp = MagicMock()
|
||||
|
||||
attacker = Player(
|
||||
name="Goku", x=0, y=0, pl=100.0, stamina=50.0, writer=mock_writer_1
|
||||
)
|
||||
defender = Player(
|
||||
name="Vegeta", x=0, y=0, pl=1.0, stamina=50.0, writer=mock_writer_2
|
||||
)
|
||||
|
||||
attacker.mode_stack.append("combat")
|
||||
defender.mode_stack.append("combat")
|
||||
|
||||
# Create encounter and attack (will kill defender)
|
||||
encounter = start_encounter(attacker, defender)
|
||||
punch = CombatMove(
|
||||
name="punch right",
|
||||
move_type="attack",
|
||||
stamina_cost=5.0,
|
||||
timing_window_ms=800,
|
||||
damage_pct=0.15,
|
||||
countered_by=["dodge left"],
|
||||
)
|
||||
encounter.attack(punch)
|
||||
|
||||
# Advance past telegraph and window to trigger resolution
|
||||
time.sleep(0.31)
|
||||
await process_combat()
|
||||
time.sleep(0.85)
|
||||
|
||||
# Reset mocks before the resolution call
|
||||
mock_writer_1.send_gmcp.reset_mock()
|
||||
mock_writer_2.send_gmcp.reset_mock()
|
||||
|
||||
await process_combat()
|
||||
|
||||
# Both players should have received Char.Status (after Char.Vitals)
|
||||
# Check that Char.Status was called at least once
|
||||
attacker_calls = [
|
||||
call
|
||||
for call in mock_writer_1.send_gmcp.call_args_list
|
||||
if call[0][0] == "Char.Status"
|
||||
]
|
||||
defender_calls = [
|
||||
call
|
||||
for call in mock_writer_2.send_gmcp.call_args_list
|
||||
if call[0][0] == "Char.Status"
|
||||
]
|
||||
|
||||
assert len(attacker_calls) >= 1, "Attacker should receive Char.Status on combat end"
|
||||
assert len(defender_calls) >= 1, "Defender should receive Char.Status on combat end"
|
||||
|
||||
# Verify mode is back to normal for attacker
|
||||
assert attacker_calls[0][0][1]["mode"] == "normal"
|
||||
assert attacker_calls[0][0][1]["in_combat"] is False
|
||||
|
||||
# Cleanup
|
||||
active_encounters.clear()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_char_status_sent_on_rest_start(player):
|
||||
"""Test Char.Status is sent when resting begins."""
|
||||
from mudlib.commands.rest import cmd_rest
|
||||
|
||||
player.stamina = 50.0
|
||||
player.resting = False
|
||||
|
||||
await cmd_rest(player, "")
|
||||
|
||||
# Check that Char.Status was sent
|
||||
status_calls = [
|
||||
call
|
||||
for call in player.writer.send_gmcp.call_args_list
|
||||
if call[0][0] == "Char.Status"
|
||||
]
|
||||
assert len(status_calls) == 1
|
||||
assert status_calls[0][0][1]["resting"] is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_char_status_sent_on_rest_complete(player):
|
||||
"""Test Char.Status is sent when resting completes."""
|
||||
from mudlib.resting import process_resting
|
||||
|
||||
player.resting = True
|
||||
player.stamina = player.max_stamina - 0.1 # Almost full
|
||||
|
||||
await process_resting()
|
||||
|
||||
# Check that Char.Status was sent
|
||||
status_calls = [
|
||||
call
|
||||
for call in player.writer.send_gmcp.call_args_list
|
||||
if call[0][0] == "Char.Status"
|
||||
]
|
||||
assert len(status_calls) == 1
|
||||
assert status_calls[0][0][1]["resting"] is False
|
||||
|
|
@ -2,6 +2,7 @@
|
|||
|
||||
import pathlib
|
||||
|
||||
from mudlib.portal import Portal
|
||||
from mudlib.zones import load_zone
|
||||
|
||||
|
||||
|
|
@ -28,7 +29,7 @@ def test_hub_zone_has_portals():
|
|||
zone = load_zone(hub_path)
|
||||
|
||||
# Find portals in zone contents
|
||||
portals = [obj for obj in zone._contents if obj.__class__.__name__ == "Portal"]
|
||||
portals = [obj for obj in zone._contents if isinstance(obj, Portal)]
|
||||
assert len(portals) == 3
|
||||
|
||||
# Check that we have portals to each target zone
|
||||
|
|
@ -43,7 +44,7 @@ def test_hub_zone_portals_at_correct_positions():
|
|||
|
||||
zone = load_zone(hub_path)
|
||||
|
||||
portals = [obj for obj in zone._contents if obj.__class__.__name__ == "Portal"]
|
||||
portals = [obj for obj in zone._contents if isinstance(obj, Portal)]
|
||||
|
||||
# North portal to overworld at (7, 0)
|
||||
overworld_portal = [p for p in portals if p.target_zone == "overworld"][0]
|
||||
|
|
|
|||
|
|
@ -42,6 +42,14 @@ class MockWriter:
|
|||
async def drain(self):
|
||||
pass
|
||||
|
||||
def send_gmcp(self, package: str, data: dict):
|
||||
"""Mock GMCP send (no-op for paint mode tests)."""
|
||||
pass
|
||||
|
||||
def send_msdp(self, data: dict):
|
||||
"""Mock MSDP send (no-op for paint mode tests)."""
|
||||
pass
|
||||
|
||||
def get_output(self) -> str:
|
||||
return "".join(self.messages)
|
||||
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
|
||||
import pathlib
|
||||
|
||||
from mudlib.portal import Portal
|
||||
from mudlib.zones import load_zone
|
||||
|
||||
|
||||
|
|
@ -59,7 +60,7 @@ def test_flower_has_portal_to_treehouse():
|
|||
zone = load_zone(flower_path)
|
||||
|
||||
# Find portals in zone contents
|
||||
portals = [obj for obj in zone._contents if obj.__class__.__name__ == "Portal"]
|
||||
portals = [obj for obj in zone._contents if isinstance(obj, Portal)]
|
||||
assert len(portals) == 1
|
||||
|
||||
portal = portals[0]
|
||||
|
|
@ -93,7 +94,7 @@ def test_treehouse_has_portal_to_overworld():
|
|||
zone = load_zone(treehouse_path)
|
||||
|
||||
# Find portals
|
||||
portals = [obj for obj in zone._contents if obj.__class__.__name__ == "Portal"]
|
||||
portals = [obj for obj in zone._contents if isinstance(obj, Portal)]
|
||||
|
||||
# Find the overworld portal
|
||||
overworld_portals = [p for p in portals if p.target_zone == "overworld"]
|
||||
|
|
@ -113,7 +114,7 @@ def test_treehouse_has_portal_to_hub():
|
|||
zone = load_zone(treehouse_path)
|
||||
|
||||
# Find portals
|
||||
portals = [obj for obj in zone._contents if obj.__class__.__name__ == "Portal"]
|
||||
portals = [obj for obj in zone._contents if isinstance(obj, Portal)]
|
||||
|
||||
# Find the hub portal
|
||||
hub_portals = [p for p in portals if p.target_zone == "hub"]
|
||||
|
|
|
|||
|
|
@ -195,9 +195,7 @@ def test_export_zone_with_portals_round_trip():
|
|||
assert loaded.height == zone.height
|
||||
|
||||
# Verify portals were loaded correctly
|
||||
portals = [
|
||||
obj for obj in loaded._contents if obj.__class__.__name__ == "Portal"
|
||||
]
|
||||
portals = [obj for obj in loaded._contents if isinstance(obj, Portal)]
|
||||
assert len(portals) == 2
|
||||
|
||||
# Sort by y coordinate for consistent ordering
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@
|
|||
import pathlib
|
||||
import tempfile
|
||||
|
||||
from mudlib.portal import Portal
|
||||
from mudlib.zones import load_zone, load_zones
|
||||
|
||||
|
||||
|
|
@ -254,7 +255,7 @@ label = "forest path"
|
|||
zone = load_zone(temp_path)
|
||||
|
||||
# Find portals in zone contents
|
||||
portals = [obj for obj in zone._contents if obj.__class__.__name__ == "Portal"]
|
||||
portals = [obj for obj in zone._contents if isinstance(obj, Portal)]
|
||||
assert len(portals) == 2
|
||||
|
||||
# Check first portal (tavern door at 5,3)
|
||||
|
|
@ -307,7 +308,7 @@ rows = [
|
|||
zone = load_zone(temp_path)
|
||||
|
||||
# Should have no portals
|
||||
portals = [obj for obj in zone._contents if obj.__class__.__name__ == "Portal"]
|
||||
portals = [obj for obj in zone._contents if isinstance(obj, Portal)]
|
||||
assert len(portals) == 0
|
||||
finally:
|
||||
temp_path.unlink()
|
||||
|
|
@ -343,7 +344,7 @@ aliases = ["gateway", "gate", "portal"]
|
|||
try:
|
||||
zone = load_zone(temp_path)
|
||||
|
||||
portals = [obj for obj in zone._contents if obj.__class__.__name__ == "Portal"]
|
||||
portals = [obj for obj in zone._contents if isinstance(obj, Portal)]
|
||||
assert len(portals) == 1
|
||||
|
||||
portal = portals[0]
|
||||
|
|
@ -361,7 +362,7 @@ def test_tavern_has_portal_to_hub():
|
|||
zone = load_zone(tavern_path)
|
||||
|
||||
# Find portals in zone
|
||||
portals = [obj for obj in zone._contents if obj.__class__.__name__ == "Portal"]
|
||||
portals = [obj for obj in zone._contents if isinstance(obj, Portal)]
|
||||
assert len(portals) == 1
|
||||
|
||||
# Check portal is at the door (4, 5) and leads to hub
|
||||
|
|
|
|||
Loading…
Reference in a new issue