File size: 3,763 Bytes
d3a24e0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | """Integration tests for the WayfinderAgent."""
from __future__ import annotations
import numpy as np
import pytest
from agents.wayfinder.agent import WayfinderAgent
class TestWayfinderAgent:
"""Integration test cases for WayfinderAgent."""
@pytest.fixture
def agent(self) -> WayfinderAgent:
"""Create a test agent with small dimensions."""
return WayfinderAgent(
max_actions=100,
latent_dim=32,
buffer_size=1000,
device="cpu",
)
@pytest.fixture
def frame(self) -> np.ndarray:
"""Create a sample frame."""
return np.random.randint(0, 16, size=(64, 64), dtype=np.uint8)
def test_act_returns_valid_action(self, agent: WayfinderAgent, frame: np.ndarray) -> None:
"""Test that act returns a valid action dict."""
result = agent.act(
frames=[frame],
state="NOT_FINISHED",
score=0.0,
win_score=1.0,
available_actions=["ACTION1", "ACTION2", "ACTION3", "ACTION4", "ACTION5"],
)
assert "action" in result
assert "reasoning" in result
assert result["action"] in ["ACTION1", "ACTION2", "ACTION3", "ACTION4", "ACTION5"]
def test_is_done_on_win(self, agent: WayfinderAgent) -> None:
"""Test that is_done returns True on WIN state."""
assert agent.is_done([], "WIN")
def test_is_done_on_game_over(self, agent: WayfinderAgent) -> None:
"""Test that is_done returns True on GAME_OVER state."""
assert agent.is_done([], "GAME_OVER")
def test_is_done_on_budget_exhausted(self, agent: WayfinderAgent) -> None:
"""Test that is_done returns True when action budget is exhausted."""
agent.action_count = agent.max_actions
assert agent.is_done([], "NOT_FINISHED")
def test_is_done_false_during_play(self, agent: WayfinderAgent) -> None:
"""Test that is_done returns False during normal play."""
agent.action_count = 5
assert not agent.is_done([], "NOT_FINISHED")
def test_reset_clears_state(self, agent: WayfinderAgent, frame: np.ndarray) -> None:
"""Test that reset clears per-level state."""
agent.act(
frames=[frame],
state="NOT_FINISHED",
score=0.0,
win_score=1.0,
available_actions=["ACTION1"],
)
assert agent.action_count > 0
agent.reset()
assert agent.action_count == 0
assert agent._prev_latent is None
def test_multiple_steps_accumulate_transitions(
self, agent: WayfinderAgent, frame: np.ndarray
) -> None:
"""Test that multiple steps build up the world model buffer."""
for i in range(5):
f = np.random.randint(0, 16, size=(64, 64), dtype=np.uint8)
agent.act(
frames=[f],
state="NOT_FINISHED",
score=0.0,
win_score=1.0,
available_actions=["ACTION1", "ACTION2", "ACTION3"],
)
# After 5 steps, at least 3 transitions should be in the buffer
assert agent._world_model.buffer_size_current >= 3
def test_reasoning_blob_has_required_fields(self, agent: WayfinderAgent, frame: np.ndarray) -> None:
"""Test that the reasoning blob contains required audit fields."""
result = agent.act(
frames=[frame],
state="NOT_FINISHED",
score=0.0,
win_score=1.0,
available_actions=["ACTION1"],
)
reasoning = result["reasoning"]
assert "step" in reasoning
assert "score" in reasoning
assert "model_confidence" in reasoning
assert "novelty" in reasoning
|