Jared Miller
42bb3fa046
Phase 5: process_mobs() runs each tick, handling mob attack and defense decisions. Mobs pick random attacks from their move list when IDLE, swap roles if needed, and attempt defense during TELEGRAPH/WINDOW with a 40% chance of correct counter. 1-second cooldown between actions. Training dummies with empty moves never fight back.
333 lines
9.6 KiB
Python
333 lines
9.6 KiB
Python
"""Tests for mob AI behavior."""
|
|
|
|
import time
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
import mudlib.commands.movement as movement_mod
|
|
from mudlib.combat import commands as combat_commands
|
|
from mudlib.combat.encounter import CombatState
|
|
from mudlib.combat.engine import (
|
|
active_encounters,
|
|
get_encounter,
|
|
start_encounter,
|
|
)
|
|
from mudlib.combat.moves import CombatMove, load_moves
|
|
from mudlib.entity import Mob
|
|
from mudlib.mob_ai import process_mobs
|
|
from mudlib.mobs import (
|
|
despawn_mob,
|
|
load_mob_template,
|
|
mobs,
|
|
spawn_mob,
|
|
)
|
|
from mudlib.player import Player, players
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_state():
|
|
"""Clear mobs, encounters, and players before and after each test."""
|
|
mobs.clear()
|
|
active_encounters.clear()
|
|
players.clear()
|
|
yield
|
|
mobs.clear()
|
|
active_encounters.clear()
|
|
players.clear()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_world():
|
|
"""Inject a mock world for movement and combat commands."""
|
|
fake_world = MagicMock()
|
|
fake_world.width = 256
|
|
fake_world.height = 256
|
|
old_movement = movement_mod.world
|
|
old_combat = combat_commands.world
|
|
movement_mod.world = fake_world
|
|
combat_commands.world = fake_world
|
|
yield fake_world
|
|
movement_mod.world = old_movement
|
|
combat_commands.world = old_combat
|
|
|
|
|
|
@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):
|
|
p = Player(
|
|
name="Goku", x=0, y=0, reader=mock_reader, writer=mock_writer
|
|
)
|
|
players[p.name] = p
|
|
return p
|
|
|
|
|
|
@pytest.fixture
|
|
def moves():
|
|
"""Load combat moves from content directory."""
|
|
content_dir = Path(__file__).parent.parent / "content" / "combat"
|
|
return load_moves(content_dir)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def inject_moves(moves):
|
|
"""Inject loaded moves into combat commands module."""
|
|
combat_commands.combat_moves = moves
|
|
yield
|
|
combat_commands.combat_moves = {}
|
|
|
|
|
|
@pytest.fixture
|
|
def goblin_toml(tmp_path):
|
|
path = tmp_path / "goblin.toml"
|
|
path.write_text(
|
|
'name = "goblin"\n'
|
|
'description = "a snarling goblin with a crude club"\n'
|
|
"pl = 50.0\n"
|
|
"stamina = 40.0\n"
|
|
"max_stamina = 40.0\n"
|
|
'moves = ["punch left", "punch right", "sweep"]\n'
|
|
)
|
|
return path
|
|
|
|
|
|
@pytest.fixture
|
|
def dummy_toml(tmp_path):
|
|
path = tmp_path / "training_dummy.toml"
|
|
path.write_text(
|
|
'name = "training dummy"\n'
|
|
'description = "a battered wooden training dummy"\n'
|
|
"pl = 200.0\n"
|
|
"stamina = 100.0\n"
|
|
"max_stamina = 100.0\n"
|
|
"moves = []\n"
|
|
)
|
|
return path
|
|
|
|
|
|
class TestMobAttackAI:
|
|
@pytest.mark.asyncio
|
|
async def test_mob_attacks_when_idle_and_cooldown_expired(
|
|
self, player, goblin_toml, moves
|
|
):
|
|
"""Mob attacks when encounter is IDLE and cooldown has expired."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0)
|
|
mob.next_action_at = 0.0 # cooldown expired
|
|
|
|
encounter = start_encounter(player, mob)
|
|
player.mode_stack.append("combat")
|
|
|
|
await process_mobs(moves)
|
|
|
|
# Mob should have attacked — encounter state should be TELEGRAPH
|
|
assert encounter.state == CombatState.TELEGRAPH
|
|
assert encounter.current_move is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mob_picks_from_its_own_moves(
|
|
self, player, goblin_toml, moves
|
|
):
|
|
"""Mob only picks moves from its moves list."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0)
|
|
mob.next_action_at = 0.0
|
|
|
|
encounter = start_encounter(player, mob)
|
|
player.mode_stack.append("combat")
|
|
|
|
await process_mobs(moves)
|
|
|
|
assert encounter.current_move is not None
|
|
assert encounter.current_move.name in mob.moves
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mob_skips_when_stamina_too_low(
|
|
self, player, goblin_toml, moves
|
|
):
|
|
"""Mob skips attack when stamina is too low for any move."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0)
|
|
mob.stamina = 0.0
|
|
mob.next_action_at = 0.0
|
|
|
|
encounter = start_encounter(player, mob)
|
|
player.mode_stack.append("combat")
|
|
|
|
await process_mobs(moves)
|
|
|
|
# Mob can't afford any move, encounter stays IDLE
|
|
assert encounter.state == CombatState.IDLE
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mob_respects_cooldown(
|
|
self, player, goblin_toml, moves
|
|
):
|
|
"""Mob doesn't act when cooldown hasn't expired."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0)
|
|
mob.next_action_at = time.monotonic() + 100.0 # far in the future
|
|
|
|
encounter = start_encounter(player, mob)
|
|
player.mode_stack.append("combat")
|
|
|
|
await process_mobs(moves)
|
|
|
|
# Mob should not have attacked
|
|
assert encounter.state == CombatState.IDLE
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mob_swaps_roles_when_defending(
|
|
self, player, goblin_toml, moves
|
|
):
|
|
"""Mob swaps attacker/defender roles when it attacks as defender."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0)
|
|
mob.next_action_at = 0.0
|
|
|
|
# Player is attacker, mob is defender
|
|
encounter = start_encounter(player, mob)
|
|
player.mode_stack.append("combat")
|
|
|
|
await process_mobs(moves)
|
|
|
|
# Mob should now be the attacker
|
|
assert encounter.attacker is mob
|
|
assert encounter.defender is player
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mob_doesnt_act_outside_combat(
|
|
self, goblin_toml, moves
|
|
):
|
|
"""Mob not in combat does nothing."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0)
|
|
mob.next_action_at = 0.0
|
|
|
|
await process_mobs(moves)
|
|
|
|
# No encounter exists for mob
|
|
assert get_encounter(mob) is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mob_sets_cooldown_after_attack(
|
|
self, player, goblin_toml, moves
|
|
):
|
|
"""Mob sets next_action_at after attacking."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0)
|
|
mob.next_action_at = 0.0
|
|
|
|
encounter = start_encounter(player, mob)
|
|
player.mode_stack.append("combat")
|
|
|
|
before = time.monotonic()
|
|
await process_mobs(moves)
|
|
|
|
# next_action_at should be ~1 second in the future
|
|
assert mob.next_action_at >= before + 0.9
|
|
|
|
|
|
class TestMobDefenseAI:
|
|
@pytest.fixture
|
|
def punch_right(self, moves):
|
|
return moves["punch right"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mob_defends_during_telegraph(
|
|
self, player, goblin_toml, moves, punch_right
|
|
):
|
|
"""Mob attempts defense during TELEGRAPH phase."""
|
|
template = load_mob_template(goblin_toml)
|
|
# Give the mob defense moves
|
|
mob = spawn_mob(template, 0, 0)
|
|
mob.moves = ["punch left", "dodge left", "dodge right"]
|
|
mob.next_action_at = 0.0
|
|
|
|
encounter = start_encounter(player, mob)
|
|
player.mode_stack.append("combat")
|
|
|
|
# Player attacks, putting encounter in TELEGRAPH
|
|
encounter.attack(punch_right)
|
|
assert encounter.state == CombatState.TELEGRAPH
|
|
|
|
await process_mobs(moves)
|
|
|
|
# Mob should have queued a defense
|
|
assert encounter.pending_defense is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mob_skips_defense_when_already_defending(
|
|
self, player, goblin_toml, moves, punch_right
|
|
):
|
|
"""Mob doesn't double-defend if already has pending_defense."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0)
|
|
mob.moves = ["dodge left", "dodge right"]
|
|
mob.next_action_at = 0.0
|
|
|
|
encounter = start_encounter(player, mob)
|
|
player.mode_stack.append("combat")
|
|
encounter.attack(punch_right)
|
|
|
|
# Pre-set a defense
|
|
existing_defense = moves["dodge left"]
|
|
encounter.pending_defense = existing_defense
|
|
|
|
await process_mobs(moves)
|
|
|
|
# Should not have changed
|
|
assert encounter.pending_defense is existing_defense
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mob_no_defense_without_defense_moves(
|
|
self, player, goblin_toml, moves, punch_right
|
|
):
|
|
"""Mob with no defense moves in its list can't defend."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0)
|
|
# Only attack moves
|
|
mob.moves = ["punch left", "punch right", "sweep"]
|
|
mob.next_action_at = time.monotonic() + 100.0 # prevent attacking
|
|
|
|
encounter = start_encounter(player, mob)
|
|
player.mode_stack.append("combat")
|
|
encounter.attack(punch_right)
|
|
|
|
await process_mobs(moves)
|
|
|
|
# No defense queued
|
|
assert encounter.pending_defense is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dummy_never_fights_back(
|
|
self, player, dummy_toml, moves, punch_right
|
|
):
|
|
"""Training dummy with empty moves never attacks or defends."""
|
|
template = load_mob_template(dummy_toml)
|
|
mob = spawn_mob(template, 0, 0)
|
|
mob.next_action_at = 0.0
|
|
|
|
encounter = start_encounter(player, mob)
|
|
player.mode_stack.append("combat")
|
|
|
|
# Player attacks
|
|
encounter.attack(punch_right)
|
|
|
|
await process_mobs(moves)
|
|
|
|
# Dummy should not have defended (empty moves list)
|
|
assert encounter.pending_defense is None
|