357 lines
11 KiB
Python
357 lines
11 KiB
Python
"""Tests for mob AI behavior."""
|
|
|
|
import time
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
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 load_moves
|
|
from mudlib.mob_ai import process_mobs
|
|
from mudlib.mobs import (
|
|
load_mob_template,
|
|
mobs,
|
|
spawn_mob,
|
|
)
|
|
from mudlib.player import Player, players
|
|
from mudlib.zone import Zone
|
|
|
|
|
|
@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
|
|
def test_zone():
|
|
"""Create a test zone for entities."""
|
|
terrain = [["." for _ in range(256)] for _ in range(256)]
|
|
zone = Zone(
|
|
name="testzone",
|
|
width=256,
|
|
height=256,
|
|
toroidal=True,
|
|
terrain=terrain,
|
|
impassable=set(),
|
|
)
|
|
return 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 player(mock_reader, mock_writer, test_zone):
|
|
p = Player(name="Goku", x=0, y=0, reader=mock_reader, writer=mock_writer)
|
|
p.location = test_zone
|
|
test_zone._contents.append(p)
|
|
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, test_zone
|
|
):
|
|
"""Mob attacks when encounter is IDLE and cooldown has expired."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0, test_zone)
|
|
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, test_zone
|
|
):
|
|
"""Mob only picks moves from its moves list."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0, test_zone)
|
|
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, test_zone
|
|
):
|
|
"""Mob skips attack when stamina is too low for any move."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0, test_zone)
|
|
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, test_zone):
|
|
"""Mob doesn't act when cooldown hasn't expired."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0, test_zone)
|
|
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, test_zone
|
|
):
|
|
"""Mob swaps attacker/defender roles when it attacks as defender."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0, test_zone)
|
|
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, test_zone):
|
|
"""Mob not in combat does nothing."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0, test_zone)
|
|
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, test_zone
|
|
):
|
|
"""Mob sets next_action_at after attacking."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0, test_zone)
|
|
mob.next_action_at = 0.0
|
|
|
|
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
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mob_sends_telegraph_to_player(
|
|
self, player, goblin_toml, moves, test_zone
|
|
):
|
|
"""When mob attacks, player receives the telegraph message."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0, test_zone)
|
|
mob.next_action_at = 0.0
|
|
|
|
start_encounter(player, mob)
|
|
player.mode_stack.append("combat")
|
|
|
|
await process_mobs(moves)
|
|
|
|
# Player's writer should have received the telegraph
|
|
# The telegraph should contain mob name and actual telegraph content
|
|
player.writer.write.assert_called()
|
|
calls = [str(call) for call in player.writer.write.call_args_list]
|
|
# Check for actual telegraph patterns from the TOML files
|
|
# Goblin moves: punch left/right ("retracts"), sweep ("drops low")
|
|
telegraph_patterns = ["retracts", "drops low"]
|
|
telegraph_sent = any(
|
|
"goblin" in call.lower()
|
|
and any(pattern in call.lower() for pattern in telegraph_patterns)
|
|
for call in calls
|
|
)
|
|
assert telegraph_sent, (
|
|
f"No telegraph with mob name and content found in: {calls}"
|
|
)
|
|
|
|
|
|
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, test_zone
|
|
):
|
|
"""Mob attempts defense during TELEGRAPH phase."""
|
|
template = load_mob_template(goblin_toml)
|
|
# Give the mob defense moves
|
|
mob = spawn_mob(template, 0, 0, test_zone)
|
|
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, test_zone
|
|
):
|
|
"""Mob doesn't double-defend if already has pending_defense."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0, test_zone)
|
|
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, test_zone
|
|
):
|
|
"""Mob with no defense moves in its list can't defend."""
|
|
template = load_mob_template(goblin_toml)
|
|
mob = spawn_mob(template, 0, 0, test_zone)
|
|
# 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, test_zone
|
|
):
|
|
"""Training dummy with empty moves never attacks or defends."""
|
|
template = load_mob_template(dummy_toml)
|
|
mob = spawn_mob(template, 0, 0, test_zone)
|
|
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
|