File size: 4,002 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 | """Tests for the memory graph module."""
from __future__ import annotations
import numpy as np
import pytest
from agents.wayfinder.memory_graph import MemoryGraph
from agents.wayfinder.perception import GRID_SIZE
class TestMemoryGraph:
"""Test cases for MemoryGraph."""
@pytest.fixture
def graph(self) -> MemoryGraph:
"""Create a test memory graph."""
return MemoryGraph()
@pytest.fixture
def sample_frame(self) -> np.ndarray:
"""Create a sample frame."""
return np.random.randint(0, 16, size=(GRID_SIZE, GRID_SIZE), dtype=np.uint8)
def test_hash_frame_stable(self, graph: MemoryGraph, sample_frame: np.ndarray) -> None:
"""Test that hashing is deterministic."""
h1 = graph.hash_frame(sample_frame)
h2 = graph.hash_frame(sample_frame)
assert h1 == h2
def test_hash_frame_different_for_different_frames(self, graph: MemoryGraph) -> None:
"""Test that different frames have different hashes."""
f1 = np.zeros((GRID_SIZE, GRID_SIZE), dtype=np.uint8)
f2 = np.ones((GRID_SIZE, GRID_SIZE), dtype=np.uint8)
assert graph.hash_frame(f1) != graph.hash_frame(f2)
def test_add_node_new(self, graph: MemoryGraph, sample_frame: np.ndarray) -> None:
"""Test adding a new node."""
h = graph.hash_frame(sample_frame)
latent = np.random.randn(64).astype(np.float32)
graph.add_node(h, latent, score=0.5)
assert h in graph.nodes
assert graph.nodes[h]["visit_count"] == 1
assert graph.nodes[h]["score"] == 0.5
def test_add_node_existing_increments_visits(self, graph: MemoryGraph, sample_frame: np.ndarray) -> None:
"""Test that re-adding a node increments its visit count."""
h = graph.hash_frame(sample_frame)
latent = np.random.randn(64).astype(np.float32)
graph.add_node(h, latent)
graph.add_node(h, latent)
graph.add_node(h, latent)
assert graph.nodes[h]["visit_count"] == 3
def test_novelty_new_node(self, graph: MemoryGraph) -> None:
"""Test novelty is 1.0 for unseen nodes."""
assert graph.novelty("nonexistent") == 1.0
def test_novelty_decreases_with_visits(self, graph: MemoryGraph, sample_frame: np.ndarray) -> None:
"""Test that novelty decreases as visit count increases."""
h = graph.hash_frame(sample_frame)
latent = np.random.randn(64).astype(np.float32)
graph.add_node(h, latent)
n1 = graph.novelty(h)
graph.add_node(h, latent)
n2 = graph.novelty(h)
assert n2 < n1
assert 0 < n2 < 1
def test_add_edge(self, graph: MemoryGraph, sample_frame: np.ndarray) -> None:
"""Test adding an edge between nodes."""
h1 = graph.hash_frame(sample_frame)
h2 = graph.hash_frame(sample_frame + 1)
latent = np.random.randn(64).astype(np.float32)
graph.add_node(h1, latent)
graph.add_node(h2, latent)
graph.add_edge(h1, h2, "ACTION1")
assert (h1, h2) in graph.edges
assert graph.edges[(h1, h2)]["action"] == "ACTION1"
assert len(graph.adjacency[h1]) == 1
def test_reset_clears_graph(self, graph: MemoryGraph, sample_frame: np.ndarray) -> None:
"""Test that reset clears all nodes and edges."""
h = graph.hash_frame(sample_frame)
latent = np.random.randn(64).astype(np.float32)
graph.add_node(h, latent)
graph.add_edge(h, h, "ACTION1")
graph.reset()
assert len(graph.nodes) == 0
assert len(graph.edges) == 0
def test_stats(self, graph: MemoryGraph, sample_frame: np.ndarray) -> None:
"""Test stats returns correct summary."""
h = graph.hash_frame(sample_frame)
latent = np.random.randn(64).astype(np.float32)
graph.add_node(h, latent)
graph.add_node(h, latent)
stats = graph.stats()
assert stats["node_count"] == 1
assert stats["avg_visits"] == 2.0
|