"""Tests for the PatternDetector and Insight formatting.""" import pytest from unittest.mock import MagicMock from reachy_mini_conversation_app.pattern_detector import ( PatternDetector, Insight, format_insights_for_prompt, ) class TestInsight: """Test the Insight dataclass.""" def test_to_dict(self): insight = Insight( pattern_type="correlation", summary="Test summary", detail="Test detail", confidence=0.75, entities=["med1", "headache"], period_days=30, ) d = insight.to_dict() assert d["pattern_type"] == "correlation" assert d["confidence"] == 0.75 assert "med1" in d["entities"] def test_default_entities_and_period(self): insight = Insight( pattern_type="test", summary="s", detail="d", confidence=0.5, ) assert insight.entities == [] assert insight.period_days == 30 class TestFormatInsightsForPrompt: """Test formatting insights for system prompt injection.""" def test_empty_insights_returns_empty(self): assert format_insights_for_prompt([]) == "" def test_single_insight_format(self): insights = [ Insight( pattern_type="correlation", summary="Headache appeared on 5 days.", detail="detail", confidence=0.8, ) ] result = format_insights_for_prompt(insights) assert "Recent Health Insights" in result assert "Correlation" in result assert "80%" in result assert "Headache appeared on 5 days." in result def test_multiple_insights_numbered(self): insights = [ Insight( pattern_type="correlation", summary="s1", detail="d1", confidence=0.9 ), Insight( pattern_type="frequency_change", summary="s2", detail="d2", confidence=0.7, ), ] result = format_insights_for_prompt(insights) assert "1." in result assert "2." in result def test_observational_language_guidance(self): """Prompt should instruct the model to use observational language.""" insights = [ Insight(pattern_type="test", summary="s", detail="d", confidence=0.5), ] result = format_insights_for_prompt(insights) assert "observational" in result.lower() or "I noticed" in result class TestPatternDetectorRunAnalysis: """Test the run_analysis orchestration.""" def test_returns_empty_when_disconnected(self): mock_graph = MagicMock() mock_graph.is_connected = False detector = PatternDetector(mock_graph) insights = detector.run_analysis("Elena", days=30) assert insights == [] def test_returns_empty_when_no_graph(self): detector = PatternDetector(None) insights = detector.run_analysis("Elena", days=30) assert insights == [] def test_continues_on_individual_detector_failure(self): """If one detector fails, the others should still run.""" mock_graph = MagicMock() mock_graph.is_connected = True # execute_read will raise on first call, return empty on subsequent mock_graph.execute_read.side_effect = [ Exception("Neo4j error"), # medication_symptom_correlation [], # frequency_changes headache [], # frequency_changes migraine [], # frequency_changes confusion [], # missed_medication_impact [], # temporal_patterns ] detector = PatternDetector(mock_graph) # Should not raise, just log warnings insights = detector.run_analysis("Elena", days=30) assert isinstance(insights, list) def test_sorts_by_confidence_descending(self): """Insights should be sorted by confidence (highest first).""" mock_graph = MagicMock() mock_graph.is_connected = True # Mock medication_symptom_correlation to return data mock_graph.execute_read.side_effect = [ # medication_symptom_correlation [ { "medication": "Med A", "symptom": "headache", "co_occurrence_count": 10, "distinct_days": 8, }, { "medication": "Med B", "symptom": "fatigue", "co_occurrence_count": 3, "distinct_days": 3, }, ], # All other detectors return empty [], [], [], [], [], ] detector = PatternDetector(mock_graph) insights = detector.run_analysis("Elena", days=30) if len(insights) >= 2: assert insights[0].confidence >= insights[1].confidence def test_caps_at_five_insights(self): """Should return at most 5 insights.""" mock_graph = MagicMock() mock_graph.is_connected = True # Return lots of correlations mock_graph.execute_read.side_effect = [ [ { "medication": f"Med{i}", "symptom": f"sym{i}", "co_occurrence_count": 5, "distinct_days": 5, } for i in range(10) ], [], [], [], [], [], ] detector = PatternDetector(mock_graph) insights = detector.run_analysis("Elena", days=30) assert len(insights) <= 5 class TestPatternDetectorInsightLanguage: """Verify insights never use causal language.""" def test_correlation_summary_is_neutral(self): mock_graph = MagicMock() mock_graph.is_connected = True mock_graph.execute_read.side_effect = [ [ { "medication": "Topiramate", "symptom": "headache", "co_occurrence_count": 5, "distinct_days": 5, } ], [], [], [], [], [], ] detector = PatternDetector(mock_graph) insights = detector.run_analysis("Elena", days=30) for insight in insights: summary_lower = insight.summary.lower() assert ( "caused" not in summary_lower ), f"Causal language in: {insight.summary}" assert ( "triggered" not in summary_lower ), f"Causal language in: {insight.summary}" assert ( "because" not in summary_lower ), f"Causal language in: {insight.summary}" class TestPatternDetectorFrequencyChanges: """Test the frequency change detector.""" def test_detects_increase(self): mock_graph = MagicMock() mock_graph.is_connected = True mock_graph.execute_read.return_value = [ {"period": "prior", "event_count": 2}, {"period": "recent", "event_count": 6}, ] detector = PatternDetector(mock_graph) insights = detector.detect_frequency_changes("Elena", "headache", days=30) assert len(insights) == 1 assert "increased" in insights[0].summary.lower() def test_detects_decrease(self): mock_graph = MagicMock() mock_graph.is_connected = True mock_graph.execute_read.return_value = [ {"period": "prior", "event_count": 10}, {"period": "recent", "event_count": 3}, ] detector = PatternDetector(mock_graph) insights = detector.detect_frequency_changes("Elena", "headache", days=30) assert len(insights) == 1 assert "decreased" in insights[0].summary.lower() def test_ignores_small_changes(self): """Changes under 25% should not generate insights.""" mock_graph = MagicMock() mock_graph.is_connected = True mock_graph.execute_read.return_value = [ {"period": "prior", "event_count": 10}, {"period": "recent", "event_count": 11}, ] detector = PatternDetector(mock_graph) insights = detector.detect_frequency_changes("Elena", "headache", days=30) assert len(insights) == 0 def test_handles_insufficient_data(self): """Should return nothing with fewer than MIN_SAMPLE_SIZE events.""" mock_graph = MagicMock() mock_graph.is_connected = True mock_graph.execute_read.return_value = [ {"period": "recent", "event_count": 1}, ] detector = PatternDetector(mock_graph) insights = detector.detect_frequency_changes("Elena", "headache", days=30) assert len(insights) == 0