142 lines
4.7 KiB
Python
142 lines
4.7 KiB
Python
|
|
"""Tests for Symbolic Memory / Intersymbolic Layer.
|
||
|
|
|
||
|
|
Generated by Allegro during PR #9 review.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from unittest.mock import MagicMock, patch
|
||
|
|
import json
|
||
|
|
|
||
|
|
|
||
|
|
class TestSymbolicMemory:
|
||
|
|
"""Test suite for agent/symbolic_memory.py"""
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def mock_adapter(self):
|
||
|
|
"""Mock GeminiAdapter."""
|
||
|
|
with patch('agent.symbolic_memory.GeminiAdapter') as MockAdapter:
|
||
|
|
mock = MagicMock()
|
||
|
|
MockAdapter.return_value = mock
|
||
|
|
yield mock
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def mock_store(self):
|
||
|
|
"""Mock GraphStore."""
|
||
|
|
with patch('agent.symbolic_memory.GraphStore') as MockStore:
|
||
|
|
mock = MagicMock()
|
||
|
|
MockStore.return_value = mock
|
||
|
|
yield mock
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def memory(self, mock_adapter, mock_store):
|
||
|
|
"""Create SymbolicMemory with mocked deps."""
|
||
|
|
from agent.symbolic_memory import SymbolicMemory
|
||
|
|
return SymbolicMemory()
|
||
|
|
|
||
|
|
def test_ingest_text_success(self, memory, mock_adapter, mock_store):
|
||
|
|
"""Should extract triples and add to graph."""
|
||
|
|
mock_adapter.generate.return_value = {
|
||
|
|
"text": json.dumps([
|
||
|
|
{"s": "Timmy", "p": "is_a", "o": "AI"},
|
||
|
|
{"s": "Timmy", "p": "has_goal", "o": "Sovereignty"}
|
||
|
|
])
|
||
|
|
}
|
||
|
|
mock_store.add_triples.return_value = 2
|
||
|
|
|
||
|
|
count = memory.ingest_text("Timmy is an AI with the goal of Sovereignty.")
|
||
|
|
|
||
|
|
assert count == 2
|
||
|
|
mock_store.add_triples.assert_called_once()
|
||
|
|
|
||
|
|
def test_ingest_text_invalid_json(self, memory, mock_adapter, mock_store):
|
||
|
|
"""Should handle malformed JSON gracefully."""
|
||
|
|
mock_adapter.generate.return_value = {
|
||
|
|
"text": "not valid json"
|
||
|
|
}
|
||
|
|
|
||
|
|
count = memory.ingest_text("Some text that confuses the model")
|
||
|
|
|
||
|
|
assert count == 0 # Should fail gracefully
|
||
|
|
mock_store.add_triples.assert_not_called()
|
||
|
|
|
||
|
|
def test_ingest_text_not_list(self, memory, mock_adapter, mock_store):
|
||
|
|
"""Should handle non-list JSON response."""
|
||
|
|
mock_adapter.generate.return_value = {
|
||
|
|
"text": json.dumps({"s": "Timmy", "p": "is_a", "o": "AI"}) # Dict, not list
|
||
|
|
}
|
||
|
|
|
||
|
|
count = memory.ingest_text("Timmy is an AI")
|
||
|
|
|
||
|
|
# Current implementation might fail here - this test documents the gap
|
||
|
|
# Should be handled: check isinstance(triples, list)
|
||
|
|
|
||
|
|
def test_get_context_for_direct_relations(self, memory, mock_store):
|
||
|
|
"""Should find direct 1-hop relations."""
|
||
|
|
mock_store.query.side_effect = lambda subject=None, **kwargs: [
|
||
|
|
{"s": "Timmy", "p": "is_a", "o": "AI"},
|
||
|
|
{"s": "Timmy", "p": "works_at", "o": "Foundation"}
|
||
|
|
] if subject == "Timmy" else []
|
||
|
|
|
||
|
|
context = memory.get_context_for("Timmy")
|
||
|
|
|
||
|
|
assert "Timmy" in context
|
||
|
|
assert "is_a" in context
|
||
|
|
assert "AI" in context
|
||
|
|
|
||
|
|
def test_get_context_for_2hop(self, memory, mock_store):
|
||
|
|
"""Should find 2-hop relations."""
|
||
|
|
# First call: direct relations
|
||
|
|
# Second call: extended relations
|
||
|
|
mock_store.query.side_effect = [
|
||
|
|
[{"s": "Timmy", "p": "works_at", "o": "Foundation"}], # Direct
|
||
|
|
[{"s": "Foundation", "p": "founded_by", "o": "Alexander"}] # 2-hop
|
||
|
|
]
|
||
|
|
|
||
|
|
context = memory.get_context_for("Timmy")
|
||
|
|
|
||
|
|
assert "Foundation" in context
|
||
|
|
assert "founded_by" in context
|
||
|
|
|
||
|
|
def test_get_context_for_empty(self, memory, mock_store):
|
||
|
|
"""Should return empty string when no context found."""
|
||
|
|
mock_store.query.return_value = []
|
||
|
|
|
||
|
|
context = memory.get_context_for("UnknownEntity")
|
||
|
|
|
||
|
|
assert context == ""
|
||
|
|
|
||
|
|
|
||
|
|
class TestIntersymbolicGraphSkill:
|
||
|
|
"""Test suite for skills/memory/intersymbolic_graph.py"""
|
||
|
|
|
||
|
|
@patch('skills.memory.intersymbolic_graph.SymbolicMemory')
|
||
|
|
def test_query_graph_with_results(self, MockMemory):
|
||
|
|
"""Skill should return formatted context."""
|
||
|
|
from skills.memory.intersymbolic_graph import query_graph
|
||
|
|
|
||
|
|
mock_instance = MagicMock()
|
||
|
|
mock_instance.get_context_for.return_value = "- Timmy --(is_a)--> AI\n"
|
||
|
|
MockMemory.return_value = mock_instance
|
||
|
|
|
||
|
|
result = query_graph("Timmy")
|
||
|
|
|
||
|
|
assert "Timmy" in result
|
||
|
|
assert "is_a" in result
|
||
|
|
|
||
|
|
@patch('skills.memory.intersymbolic_graph.SymbolicMemory')
|
||
|
|
def test_query_graph_no_results(self, MockMemory):
|
||
|
|
"""Skill should handle empty results gracefully."""
|
||
|
|
from skills.memory.intersymbolic_graph import query_graph
|
||
|
|
|
||
|
|
mock_instance = MagicMock()
|
||
|
|
mock_instance.get_context_for.return_value = ""
|
||
|
|
MockMemory.return_value = mock_instance
|
||
|
|
|
||
|
|
result = query_graph("Unknown")
|
||
|
|
|
||
|
|
assert "No symbolic connections" in result
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
pytest.main([__file__, "-v"])
|