Spaces:
Running
Running
| """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() | |
| 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 | |