Mobs with home regions now pathfind back when they've strayed. Each tick, process_mob_movement() checks all mobs and moves them one tile toward their home region center using Manhattan distance. Movement is throttled to 3 seconds, respects impassable terrain, skips mobs in combat, and broadcasts to nearby players.
259 lines
6 KiB
Python
259 lines
6 KiB
Python
"""Tests for mob pathfinding back to home region."""
|
|
|
|
import time
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from mudlib.entity import Mob
|
|
from mudlib.mob_ai import process_mob_movement
|
|
from mudlib.mobs import mobs
|
|
from mudlib.zone import Zone
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_mobs():
|
|
"""Clear global mobs list before each test."""
|
|
mobs.clear()
|
|
yield
|
|
mobs.clear()
|
|
|
|
|
|
@pytest.fixture
|
|
def zone():
|
|
"""Create a simple test zone."""
|
|
terrain = [
|
|
[".", ".", ".", ".", "."],
|
|
[".", ".", "#", ".", "."],
|
|
[".", ".", ".", ".", "."],
|
|
[".", ".", "#", ".", "."],
|
|
[".", ".", ".", ".", "."],
|
|
]
|
|
return Zone(
|
|
name="test",
|
|
description="Test Zone",
|
|
width=5,
|
|
height=5,
|
|
terrain=terrain,
|
|
toroidal=False,
|
|
impassable={"#"},
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mob_outside_home_region_moves_toward_it(zone):
|
|
"""Mob outside home region should move one tile toward center."""
|
|
# Create mob at (4, 4) with home region center at (1, 1)
|
|
mob = Mob(
|
|
name="test_mob",
|
|
location=zone,
|
|
x=4,
|
|
y=4,
|
|
home_x_min=0,
|
|
home_x_max=2,
|
|
home_y_min=0,
|
|
home_y_max=2,
|
|
)
|
|
mobs.append(mob)
|
|
|
|
# First movement should happen immediately
|
|
await process_mob_movement()
|
|
|
|
# Mob should move toward home (prefer larger distance axis)
|
|
# Distance to center (1,1): dx=3, dy=3 — equal, so prefer x-axis
|
|
assert mob.x == 3 and mob.y == 4
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mob_inside_home_region_does_not_move(zone):
|
|
"""Mob already inside home region should not move."""
|
|
mob = Mob(
|
|
name="test_mob",
|
|
location=zone,
|
|
x=1,
|
|
y=1,
|
|
home_x_min=0,
|
|
home_x_max=2,
|
|
home_y_min=0,
|
|
home_y_max=2,
|
|
)
|
|
mobs.append(mob)
|
|
|
|
await process_mob_movement()
|
|
|
|
# Position should not change
|
|
assert mob.x == 1 and mob.y == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mob_with_no_home_region_does_not_move(zone):
|
|
"""Mob with no home region should not move."""
|
|
mob = Mob(
|
|
name="test_mob",
|
|
location=zone,
|
|
x=2,
|
|
y=2,
|
|
)
|
|
mobs.append(mob)
|
|
|
|
await process_mob_movement()
|
|
|
|
# Position should not change
|
|
assert mob.x == 2 and mob.y == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mob_in_combat_does_not_move(zone):
|
|
"""Mob in an encounter should not move."""
|
|
from mudlib.combat.encounter import CombatEncounter
|
|
|
|
mob = Mob(
|
|
name="test_mob",
|
|
location=zone,
|
|
x=4,
|
|
y=4,
|
|
home_x_min=0,
|
|
home_x_max=2,
|
|
home_y_min=0,
|
|
home_y_max=2,
|
|
)
|
|
mobs.append(mob)
|
|
|
|
# Create a mock player for the encounter
|
|
from mudlib.player import Player
|
|
|
|
mock_writer = MagicMock()
|
|
mock_writer.is_closing.return_value = False
|
|
mock_writer.drain = AsyncMock()
|
|
|
|
player = Player(
|
|
name="test_player",
|
|
location=zone,
|
|
x=4,
|
|
y=4,
|
|
writer=mock_writer,
|
|
reader=MagicMock(),
|
|
)
|
|
|
|
# Create encounter
|
|
from mudlib.combat.engine import active_encounters
|
|
|
|
encounter = CombatEncounter(attacker=mob, defender=player)
|
|
active_encounters.append(encounter)
|
|
|
|
try:
|
|
await process_mob_movement()
|
|
|
|
# Mob should not move
|
|
assert mob.x == 4 and mob.y == 4
|
|
finally:
|
|
active_encounters.clear()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_movement_is_throttled(zone):
|
|
"""Mob should not move on every tick (throttle ~3 seconds)."""
|
|
mob = Mob(
|
|
name="test_mob",
|
|
location=zone,
|
|
x=4,
|
|
y=4,
|
|
home_x_min=0,
|
|
home_x_max=2,
|
|
home_y_min=0,
|
|
home_y_max=2,
|
|
)
|
|
mobs.append(mob)
|
|
|
|
# First move should happen
|
|
await process_mob_movement()
|
|
first_x, first_y = mob.x, mob.y
|
|
assert (first_x, first_y) != (4, 4) # Moved
|
|
|
|
# Immediate second call should not move
|
|
await process_mob_movement()
|
|
assert mob.x == first_x and mob.y == first_y
|
|
|
|
# After waiting 3 seconds, should move again
|
|
time.sleep(3.1)
|
|
await process_mob_movement()
|
|
assert mob.x != first_x or mob.y != first_y
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mob_respects_impassable_terrain(zone):
|
|
"""Mob should not move into impassable terrain."""
|
|
# Position mob at (1, 0) with home at (3, 0)
|
|
# There's a wall at (2, 1) but path around it
|
|
mob = Mob(
|
|
name="test_mob",
|
|
location=zone,
|
|
x=1,
|
|
y=0,
|
|
home_x_min=3,
|
|
home_x_max=4,
|
|
home_y_min=0,
|
|
home_y_max=1,
|
|
)
|
|
mobs.append(mob)
|
|
|
|
await process_mob_movement()
|
|
|
|
# Mob should move toward home (east)
|
|
assert mob.x == 2 and mob.y == 0
|
|
|
|
# Try to move south from (2,0) — wall at (2,1)
|
|
mob.x = 2
|
|
mob.y = 0
|
|
mob.home_x_min = 2
|
|
mob.home_x_max = 2
|
|
mob.home_y_min = 2
|
|
mob.home_y_max = 4
|
|
time.sleep(3.1) # Wait for throttle
|
|
|
|
await process_mob_movement()
|
|
|
|
# Should not have moved into wall (2,1 is '#')
|
|
# Should stay at (2, 0) since target (2,1) is blocked
|
|
assert mob.x == 2 and mob.y == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_movement_broadcasts_to_nearby_players(zone):
|
|
"""Mob movement should broadcast to nearby players."""
|
|
from mudlib.player import Player
|
|
|
|
mob = Mob(
|
|
name="test_mob",
|
|
location=zone,
|
|
x=2,
|
|
y=2,
|
|
home_x_min=0,
|
|
home_x_max=1,
|
|
home_y_min=0,
|
|
home_y_max=1,
|
|
)
|
|
mobs.append(mob)
|
|
|
|
# Create a player nearby
|
|
mock_writer = MagicMock()
|
|
mock_writer.is_closing.return_value = False
|
|
mock_writer.drain = AsyncMock()
|
|
|
|
player = Player(
|
|
name="test_player",
|
|
location=zone,
|
|
x=2,
|
|
y=2,
|
|
writer=mock_writer,
|
|
reader=MagicMock(),
|
|
)
|
|
|
|
await process_mob_movement()
|
|
|
|
# Player should receive a movement message
|
|
assert player.writer.write.called
|
|
# Check that message contains mob name and direction
|
|
call_args = [call[0][0] for call in player.writer.write.call_args_list]
|
|
movement_msg = "".join(call_args)
|
|
assert "test_mob" in movement_msg
|