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