304 lines
9.2 KiB
Python
304 lines
9.2 KiB
Python
"""Tests for three-beat combat output system."""
|
|
|
|
import time
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from mudlib.caps import ClientCaps
|
|
from mudlib.combat.encounter import CombatState
|
|
from mudlib.combat.engine import (
|
|
active_encounters,
|
|
process_combat,
|
|
start_encounter,
|
|
)
|
|
from mudlib.combat.moves import CombatMove
|
|
from mudlib.player import Player
|
|
|
|
|
|
def _mock_writer():
|
|
writer = MagicMock()
|
|
writer.write = MagicMock()
|
|
writer.drain = AsyncMock()
|
|
return writer
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_encounters():
|
|
"""Clear encounters before and after each test."""
|
|
active_encounters.clear()
|
|
yield
|
|
active_encounters.clear()
|
|
|
|
|
|
@pytest.fixture
|
|
def punch():
|
|
return CombatMove(
|
|
name="punch left",
|
|
move_type="attack",
|
|
stamina_cost=5.0,
|
|
hit_time_ms=800,
|
|
damage_pct=0.15,
|
|
countered_by=["dodge right"],
|
|
telegraph="{attacker} retracts {his} left arm...",
|
|
announce="{attacker} throw{s} a left hook at {defender}!",
|
|
resolve_hit="{attacker} connect{s} with {his} left hook!",
|
|
resolve_miss="{defender} counter{s} the left hook!",
|
|
telegraph_color="dim",
|
|
announce_color="",
|
|
resolve_color="bold",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_announce_sent_on_pending_to_resolve_transition(punch):
|
|
"""Test announce message sent when PENDING→RESOLVE transition occurs."""
|
|
atk_writer = _mock_writer()
|
|
def_writer = _mock_writer()
|
|
attacker = Player(name="Goku", x=0, y=0, pl=100.0, stamina=50.0, writer=atk_writer)
|
|
defender = Player(
|
|
name="Vegeta", x=0, y=0, pl=100.0, stamina=50.0, writer=def_writer
|
|
)
|
|
|
|
encounter = start_encounter(attacker, defender)
|
|
encounter.attack(punch)
|
|
|
|
# Should be in PENDING state
|
|
assert encounter.state == CombatState.PENDING
|
|
|
|
# Reset mocks to ignore previous messages
|
|
atk_writer.write.reset_mock()
|
|
def_writer.write.reset_mock()
|
|
|
|
# Wait for hit time (800ms) and process
|
|
time.sleep(0.85)
|
|
await process_combat()
|
|
|
|
# Should have auto-resolved and returned to IDLE
|
|
assert encounter.state == CombatState.IDLE
|
|
|
|
# Check announce message was sent
|
|
atk_msgs = [call[0][0] for call in atk_writer.write.call_args_list]
|
|
def_msgs = [call[0][0] for call in def_writer.write.call_args_list]
|
|
|
|
# Attacker should see "You throw a left hook at Vegeta!"
|
|
assert any("throw" in msg.lower() and "hook" in msg.lower() for msg in atk_msgs)
|
|
# Defender should see "Goku throws a left hook at you!"
|
|
assert any("throws" in msg.lower() and "hook" in msg.lower() for msg in def_msgs)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_announce_uses_pov_templates(punch):
|
|
"""Test announce messages use POV templates correctly."""
|
|
atk_writer = _mock_writer()
|
|
def_writer = _mock_writer()
|
|
attacker = Player(name="Goku", x=0, y=0, pl=100.0, stamina=50.0, writer=atk_writer)
|
|
defender = Player(
|
|
name="Vegeta", x=0, y=0, pl=100.0, stamina=50.0, writer=def_writer
|
|
)
|
|
|
|
encounter = start_encounter(attacker, defender)
|
|
encounter.attack(punch)
|
|
|
|
atk_writer.write.reset_mock()
|
|
def_writer.write.reset_mock()
|
|
|
|
# Wait for hit time (800ms) to trigger announce
|
|
time.sleep(0.85)
|
|
await process_combat()
|
|
|
|
atk_msgs = [call[0][0] for call in atk_writer.write.call_args_list]
|
|
def_msgs = [call[0][0] for call in def_writer.write.call_args_list]
|
|
|
|
# Attacker POV: "You throw a left hook at Vegeta!"
|
|
assert any("You throw" in msg for msg in atk_msgs)
|
|
# Defender POV: "Goku throws a left hook at You!"
|
|
assert any("Goku throws" in msg and "at You" in msg for msg in def_msgs)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_uses_pov_templates(punch):
|
|
"""Test resolve messages use POV templates correctly."""
|
|
atk_writer = _mock_writer()
|
|
def_writer = _mock_writer()
|
|
attacker = Player(name="Goku", x=0, y=0, pl=100.0, stamina=50.0, writer=atk_writer)
|
|
defender = Player(
|
|
name="Vegeta", x=0, y=0, pl=100.0, stamina=50.0, writer=def_writer
|
|
)
|
|
|
|
encounter = start_encounter(attacker, defender)
|
|
encounter.attack(punch)
|
|
|
|
# Advance to resolve (announce and resolve happen on same tick)
|
|
time.sleep(0.85)
|
|
await process_combat()
|
|
|
|
atk_msgs = [call[0][0] for call in atk_writer.write.call_args_list]
|
|
def_msgs = [call[0][0] for call in def_writer.write.call_args_list]
|
|
|
|
# Attacker POV: "You connect with your left hook!"
|
|
assert any("You connect" in msg and "your" in msg for msg in atk_msgs)
|
|
# Defender POV: "Goku connects with his left hook!"
|
|
assert any("Goku connects" in msg and "his" in msg for msg in def_msgs)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_uses_resolve_miss_on_counter(punch):
|
|
"""Test resolve uses resolve_miss template when countered."""
|
|
atk_writer = _mock_writer()
|
|
def_writer = _mock_writer()
|
|
attacker = Player(name="Goku", x=0, y=0, pl=100.0, stamina=50.0, writer=atk_writer)
|
|
defender = Player(
|
|
name="Vegeta", x=0, y=0, pl=100.0, stamina=50.0, writer=def_writer
|
|
)
|
|
|
|
dodge = CombatMove(
|
|
name="dodge right",
|
|
move_type="defense",
|
|
stamina_cost=3.0,
|
|
active_ms=1000, # Longer than hit_time to still be active at resolve
|
|
recovery_ms=2700,
|
|
)
|
|
|
|
encounter = start_encounter(attacker, defender)
|
|
encounter.attack(punch)
|
|
encounter.defend(dodge)
|
|
|
|
atk_writer.write.reset_mock()
|
|
def_writer.write.reset_mock()
|
|
|
|
# Advance to resolve (defense is already active)
|
|
time.sleep(0.85)
|
|
await process_combat()
|
|
|
|
atk_msgs = [call[0][0] for call in atk_writer.write.call_args_list]
|
|
def_msgs = [call[0][0] for call in def_writer.write.call_args_list]
|
|
|
|
# Should use resolve_miss template
|
|
# Attacker POV: "Vegeta counters your left hook!"
|
|
assert any("counter" in msg.lower() for msg in atk_msgs)
|
|
# Defender POV: "You counter Goku's left hook!"
|
|
assert any("You counter" in msg for msg in def_msgs)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_announce_color_applied(punch):
|
|
"""Test announce messages have no color wrap (default)."""
|
|
atk_writer = _mock_writer()
|
|
def_writer = _mock_writer()
|
|
# Set color depth to enable colors via caps
|
|
attacker = Player(
|
|
name="Goku",
|
|
x=0,
|
|
y=0,
|
|
pl=100.0,
|
|
stamina=50.0,
|
|
writer=atk_writer,
|
|
caps=ClientCaps(ansi=True),
|
|
)
|
|
defender = Player(
|
|
name="Vegeta",
|
|
x=0,
|
|
y=0,
|
|
pl=100.0,
|
|
stamina=50.0,
|
|
writer=def_writer,
|
|
caps=ClientCaps(ansi=True),
|
|
)
|
|
|
|
encounter = start_encounter(attacker, defender)
|
|
encounter.attack(punch)
|
|
|
|
atk_writer.write.reset_mock()
|
|
def_writer.write.reset_mock()
|
|
|
|
# Wait for hit time to trigger announce
|
|
time.sleep(0.85)
|
|
await process_combat()
|
|
|
|
atk_msgs = [call[0][0] for call in atk_writer.write.call_args_list]
|
|
|
|
# Announce has announce_color="" so should NOT have ANSI codes
|
|
# (just plain text)
|
|
announce_msgs = [msg for msg in atk_msgs if "throw" in msg.lower()]
|
|
assert len(announce_msgs) > 0
|
|
# Should not contain ANSI escape sequences for colors
|
|
# (no \033[ sequences for colors, may have for other reasons)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_bold_color_applied(punch):
|
|
"""Test resolve messages have bold color applied."""
|
|
atk_writer = _mock_writer()
|
|
def_writer = _mock_writer()
|
|
attacker = Player(
|
|
name="Goku",
|
|
x=0,
|
|
y=0,
|
|
pl=100.0,
|
|
stamina=50.0,
|
|
writer=atk_writer,
|
|
caps=ClientCaps(ansi=True),
|
|
)
|
|
defender = Player(
|
|
name="Vegeta",
|
|
x=0,
|
|
y=0,
|
|
pl=100.0,
|
|
stamina=50.0,
|
|
writer=def_writer,
|
|
caps=ClientCaps(ansi=True),
|
|
)
|
|
|
|
encounter = start_encounter(attacker, defender)
|
|
encounter.attack(punch)
|
|
|
|
atk_writer.write.reset_mock()
|
|
def_writer.write.reset_mock()
|
|
|
|
# Advance to resolve
|
|
time.sleep(0.85)
|
|
await process_combat()
|
|
|
|
atk_msgs = [call[0][0] for call in atk_writer.write.call_args_list]
|
|
|
|
# Resolve has resolve_color="bold" so should have ANSI bold codes
|
|
resolve_msgs = [msg for msg in atk_msgs if "connect" in msg.lower()]
|
|
assert len(resolve_msgs) > 0
|
|
# Should contain ANSI bold sequence
|
|
assert any("\033[1m" in msg for msg in resolve_msgs)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_announce_on_idle_to_pending():
|
|
"""Test no announce message sent on IDLE→PENDING transition."""
|
|
atk_writer = _mock_writer()
|
|
def_writer = _mock_writer()
|
|
attacker = Player(name="Goku", x=0, y=0, pl=100.0, stamina=50.0, writer=atk_writer)
|
|
defender = Player(
|
|
name="Vegeta", x=0, y=0, pl=100.0, stamina=50.0, writer=def_writer
|
|
)
|
|
|
|
punch = CombatMove(
|
|
name="punch left",
|
|
move_type="attack",
|
|
stamina_cost=5.0,
|
|
hit_time_ms=800,
|
|
damage_pct=0.15,
|
|
announce="{attacker} throw{s} a left hook at {defender}!",
|
|
)
|
|
|
|
encounter = start_encounter(attacker, defender)
|
|
|
|
atk_writer.write.reset_mock()
|
|
def_writer.write.reset_mock()
|
|
|
|
encounter.attack(punch)
|
|
|
|
# Process combat immediately (still in PENDING)
|
|
await process_combat()
|
|
|
|
atk_msgs = [call[0][0] for call in atk_writer.write.call_args_list]
|
|
|
|
# Should NOT see announce message yet
|
|
assert not any("throw" in msg.lower() for msg in atk_msgs)
|