Jared Miller
a4c9f31056
When combat begins, any active power-up task on either the attacker or defender should be cancelled to prevent background power changes during combat. This ensures players can't continue charging while fighting. The fix checks both entities for a _power_task attribute and cancels it if present, then clears the reference.
357 lines
9.8 KiB
Python
357 lines
9.8 KiB
Python
"""Tests for power commands."""
|
|
|
|
import asyncio
|
|
import time
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from mudlib.combat.encounter import CombatEncounter
|
|
from mudlib.combat.engine import active_encounters
|
|
from mudlib.commands.power import cmd_power
|
|
from mudlib.player import Player, players
|
|
from mudlib.zone import Zone
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_state():
|
|
"""Clear players and encounters before and after each test."""
|
|
players.clear()
|
|
active_encounters.clear()
|
|
yield
|
|
players.clear()
|
|
active_encounters.clear()
|
|
|
|
|
|
@pytest.fixture
|
|
def test_zone():
|
|
"""Create a test zone for players."""
|
|
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 nearby_player(mock_reader, mock_writer, test_zone):
|
|
p = Player(name="Vegeta", 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.mark.asyncio
|
|
async def test_power_down_instantly_lowers_pl(player):
|
|
"""Test power down instantly lowers PL to minimum."""
|
|
player.pl = 100.0
|
|
player.max_pl = 100.0
|
|
|
|
await cmd_power(player, "down")
|
|
|
|
assert player.pl == 1.0
|
|
messages = [call[0][0] for call in player.writer.write.call_args_list]
|
|
assert any("power down" in msg.lower() for msg in messages)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_power_to_number_lowers_instantly(player):
|
|
"""Test power <number> instantly lowers when target < current."""
|
|
player.pl = 100.0
|
|
player.max_pl = 100.0
|
|
|
|
await cmd_power(player, "50")
|
|
|
|
assert player.pl == 50.0
|
|
messages = [call[0][0] for call in player.writer.write.call_args_list]
|
|
assert any("50" in msg for msg in messages)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_power_to_number_rejects_zero(player):
|
|
"""Test power <number> rejects 0 or negative values."""
|
|
player.pl = 100.0
|
|
|
|
await cmd_power(player, "0")
|
|
|
|
assert player.pl == 100.0 # unchanged
|
|
messages = [call[0][0] for call in player.writer.write.call_args_list]
|
|
assert any("must be greater than 0" in msg.lower() for msg in messages)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_power_to_number_rejects_above_max(player):
|
|
"""Test power <number> rejects values above max_pl."""
|
|
player.pl = 100.0
|
|
player.max_pl = 100.0
|
|
|
|
await cmd_power(player, "150")
|
|
|
|
assert player.pl == 100.0 # unchanged
|
|
messages = [call[0][0] for call in player.writer.write.call_args_list]
|
|
assert any("maximum" in msg.lower() for msg in messages)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_power_up_starts_increasing_pl(player):
|
|
"""Test power up starts a loop that increases PL over time."""
|
|
player.pl = 10.0
|
|
player.max_pl = 100.0
|
|
player.stamina = 100.0
|
|
|
|
await cmd_power(player, "up")
|
|
|
|
# Give it a moment to run a few ticks
|
|
await asyncio.sleep(0.15)
|
|
|
|
assert player.pl > 10.0
|
|
assert player.pl < player.max_pl
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_power_up_deducts_stamina(player):
|
|
"""Test power up deducts stamina per tick."""
|
|
player.pl = 10.0
|
|
player.max_pl = 100.0
|
|
player.stamina = 100.0
|
|
|
|
await cmd_power(player, "up")
|
|
|
|
# Give it a moment to run a few ticks
|
|
await asyncio.sleep(0.15)
|
|
|
|
assert player.stamina < 100.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_power_up_stops_at_max_pl(player):
|
|
"""Test power up stops when PL reaches max_pl."""
|
|
player.pl = 95.0
|
|
player.max_pl = 100.0
|
|
player.stamina = 100.0
|
|
|
|
await cmd_power(player, "up")
|
|
|
|
# Wait for it to complete
|
|
await asyncio.sleep(0.3)
|
|
|
|
assert player.pl == 100.0
|
|
assert player._power_task is None or player._power_task.done()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_power_up_stops_when_stamina_depleted(player):
|
|
"""Test power up stops when stamina runs out."""
|
|
player.pl = 10.0
|
|
player.max_pl = 100.0
|
|
player.stamina = 5.0 # Very low stamina
|
|
|
|
await cmd_power(player, "up")
|
|
|
|
# Wait for stamina to deplete
|
|
await asyncio.sleep(0.3)
|
|
|
|
# PL should have increased a little, but stopped
|
|
assert player.pl > 10.0
|
|
assert player.pl < 100.0
|
|
assert player.stamina <= 0.0
|
|
assert player._power_task is None or player._power_task.done()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_power_stop_cancels_power_up(player):
|
|
"""Test power stop cancels an ongoing power-up."""
|
|
player.pl = 10.0
|
|
player.max_pl = 100.0
|
|
player.stamina = 100.0
|
|
|
|
await cmd_power(player, "up")
|
|
await asyncio.sleep(0.05) # Let it start
|
|
|
|
await cmd_power(player, "stop")
|
|
|
|
pl_at_stop = player.pl
|
|
await asyncio.sleep(0.1) # Wait a bit more
|
|
|
|
# PL should not have increased after stop
|
|
assert player.pl == pl_at_stop
|
|
assert player._power_task is None or player._power_task.done()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_power_up_rejects_during_combat(player, nearby_player):
|
|
"""Test power up is rejected when player is in combat."""
|
|
player.pl = 50.0
|
|
player.stamina = 100.0
|
|
|
|
# Put player in combat
|
|
encounter = CombatEncounter(
|
|
attacker=player, defender=nearby_player, last_action_at=time.monotonic()
|
|
)
|
|
active_encounters.append(encounter)
|
|
|
|
await cmd_power(player, "up")
|
|
|
|
messages = [call[0][0] for call in player.writer.write.call_args_list]
|
|
assert any("combat" in msg.lower() for msg in messages)
|
|
assert player.pl == 50.0 # unchanged
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_power_up_rejects_with_no_stamina(player):
|
|
"""Test power up is rejected when stamina is 0."""
|
|
player.pl = 50.0
|
|
player.stamina = 0.0
|
|
|
|
await cmd_power(player, "up")
|
|
|
|
messages = [call[0][0] for call in player.writer.write.call_args_list]
|
|
assert any("stamina" in msg.lower() for msg in messages)
|
|
assert player.pl == 50.0 # unchanged
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_power_up_sends_feedback_messages(player):
|
|
"""Test power up sends feedback to player."""
|
|
player.pl = 10.0
|
|
player.max_pl = 100.0
|
|
player.stamina = 100.0
|
|
|
|
await cmd_power(player, "up")
|
|
|
|
messages = [call[0][0] for call in player.writer.write.call_args_list]
|
|
assert any("powering up" in msg.lower() for msg in messages)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_power_up_broadcasts_to_nearby(player, nearby_player):
|
|
"""Test power up broadcasts to nearby players."""
|
|
player.pl = 10.0
|
|
player.max_pl = 100.0
|
|
player.stamina = 100.0
|
|
|
|
await cmd_power(player, "up")
|
|
|
|
nearby_messages = [call[0][0] for call in nearby_player.writer.write.call_args_list]
|
|
assert any("Goku" in msg and "power" in msg.lower() for msg in nearby_messages)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_power_up_when_already_powering_up(player):
|
|
"""Test power up when already powering up sends message."""
|
|
player.pl = 10.0
|
|
player.max_pl = 100.0
|
|
player.stamina = 100.0
|
|
|
|
await cmd_power(player, "up")
|
|
await asyncio.sleep(0.05)
|
|
|
|
# Try to power up again
|
|
await cmd_power(player, "up")
|
|
|
|
messages = [call[0][0] for call in player.writer.write.call_args_list]
|
|
assert any("already" in msg.lower() for msg in messages)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_power_down_broadcasts_to_nearby(player, nearby_player):
|
|
"""Test power down broadcasts to nearby players."""
|
|
player.pl = 100.0
|
|
|
|
await cmd_power(player, "down")
|
|
|
|
nearby_messages = [call[0][0] for call in nearby_player.writer.write.call_args_list]
|
|
assert any("Goku" in msg and "power" in msg.lower() for msg in nearby_messages)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_power_to_number_raises_with_loop(player):
|
|
"""Test power <number> starts power-up loop when target > current."""
|
|
player.pl = 10.0
|
|
player.max_pl = 100.0
|
|
player.stamina = 100.0
|
|
|
|
await cmd_power(player, "80")
|
|
|
|
# Give it a moment to run
|
|
await asyncio.sleep(0.15)
|
|
|
|
# Should be increasing toward 80
|
|
assert player.pl > 10.0
|
|
assert player.stamina < 100.0 # deducting stamina
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_combat_start_cancels_power_up(player, nearby_player):
|
|
"""Test starting combat cancels any active power-up task."""
|
|
from mudlib.combat.engine import start_encounter
|
|
|
|
# Start player powering up
|
|
player.pl = 10.0
|
|
player.max_pl = 100.0
|
|
player.stamina = 100.0
|
|
|
|
await cmd_power(player, "up")
|
|
await asyncio.sleep(0.05) # Let power-up start
|
|
|
|
# Verify power-up is active
|
|
assert player._power_task is not None
|
|
assert not player._power_task.done()
|
|
|
|
# Start combat encounter
|
|
start_encounter(player, nearby_player)
|
|
|
|
# Power-up task should be cancelled
|
|
assert player._power_task is None or player._power_task.cancelled()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_combat_start_cancels_defender_power_up(player, nearby_player):
|
|
"""Test starting combat cancels defender's active power-up task."""
|
|
from mudlib.combat.engine import start_encounter
|
|
|
|
# Start nearby_player powering up
|
|
nearby_player.pl = 10.0
|
|
nearby_player.max_pl = 100.0
|
|
nearby_player.stamina = 100.0
|
|
|
|
await cmd_power(nearby_player, "up")
|
|
await asyncio.sleep(0.05) # Let power-up start
|
|
|
|
# Verify power-up is active
|
|
assert nearby_player._power_task is not None
|
|
assert not nearby_player._power_task.done()
|
|
|
|
# Start combat encounter (nearby_player is defender)
|
|
start_encounter(player, nearby_player)
|
|
|
|
# Power-up task should be cancelled
|
|
assert nearby_player._power_task is None or nearby_player._power_task.cancelled()
|