Spaces:
Running
Running
File size: 4,048 Bytes
af9cde9 |
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 106 107 108 109 110 111 112 113 114 115 116 117 |
"""Tests for the Neo4j GraphMemory module."""
import pytest
from unittest.mock import MagicMock, patch
from reachy_mini_conversation_app.memory_graph import GraphMemory, HAS_NEO4J
class TestGraphMemoryWithoutNeo4j:
"""Test GraphMemory graceful degradation without Neo4j."""
def test_graceful_degradation_no_connection(self):
"""Test that GraphMemory works without Neo4j connection."""
graph = GraphMemory()
# Should not raise, just return False
connected = graph.connect()
# Connection will fail since Neo4j isn't running
# But we should be able to call methods without crashing
context = graph.get_patient_context("Elena")
assert context["patient"] == "Elena"
assert context["medications"] == []
graph.close()
def test_format_empty_context(self):
"""Test formatting context when no data exists."""
graph = GraphMemory()
prompt = graph.format_context_for_prompt("Elena")
assert prompt == "" # Should be empty when not connected
graph.close()
@pytest.mark.skipif(not HAS_NEO4J, reason="neo4j package not installed")
class TestGraphMemoryWithMockedDriver:
"""Test GraphMemory with mocked Neo4j driver."""
def test_add_person(self):
"""Test adding a person node."""
graph = GraphMemory()
with patch.object(graph, "_execute") as mock_execute:
graph._driver = MagicMock() # Pretend we're connected
graph.add_person("Elena", "patient")
mock_execute.assert_called_once()
call_args = mock_execute.call_args
assert "Person" in call_args[0][0]
assert call_args[0][1]["name"] == "Elena"
assert call_args[0][1]["role"] == "patient"
def test_add_medication(self):
"""Test adding a medication node."""
graph = GraphMemory()
with patch.object(graph, "_execute") as mock_execute:
graph._driver = MagicMock()
graph.add_medication(
"Topiramate",
dose="50mg",
frequency="daily",
symptom_category="migraine",
)
mock_execute.assert_called_once()
call_args = mock_execute.call_args
assert "Medication" in call_args[0][0]
assert call_args[0][1]["name"] == "Topiramate"
def test_link_patient_takes_medication(self):
"""Test creating TAKES relationship."""
graph = GraphMemory()
with patch.object(graph, "_execute") as mock_execute:
graph._driver = MagicMock()
graph.link_patient_takes_medication("Elena", "Topiramate")
mock_execute.assert_called_once()
call_args = mock_execute.call_args
assert "TAKES" in call_args[0][0]
def test_get_patient_context(self):
"""Test retrieving patient context."""
graph = GraphMemory()
with patch.object(graph, "_execute") as mock_execute:
graph._driver = MagicMock()
mock_execute.return_value = [] # Empty results
context = graph.get_patient_context("Elena")
assert context["patient"] == "Elena"
assert "medications" in context
assert "recent_events" in context
assert "caregivers" in context
def test_get_caregiver_summary(self):
"""Test generating caregiver summary for Maya."""
graph = GraphMemory()
with patch.object(graph, "_execute") as mock_execute:
graph._driver = MagicMock()
mock_execute.side_effect = [
[{"event_type": "migraine_episode", "count": 3}], # event_counts
[{"missed_count": 1}], # missed medications
[], # concerning events
]
summary = graph.get_caregiver_summary("Elena", days=7)
assert summary["patient"] == "Elena"
assert summary["period_days"] == 7
assert summary["missed_medications"] == 1
|