49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from ea_chatbot.graph.nodes.planner import planner_node
|
|
from ea_chatbot.schemas import ChecklistResponse, ChecklistTask
|
|
|
|
@pytest.fixture
|
|
def mock_state():
|
|
return {
|
|
"messages": [],
|
|
"question": "Show me results for New Jersey",
|
|
"analysis": {
|
|
"expert": "Data Analyst",
|
|
"data": "NJ data",
|
|
"unknown": "results",
|
|
"condition": "state=NJ"
|
|
},
|
|
"next_action": "plan",
|
|
"checklist": [],
|
|
"current_step": 0
|
|
}
|
|
|
|
@patch("ea_chatbot.graph.nodes.planner.get_llm_model")
|
|
@patch("ea_chatbot.utils.database_inspection.get_data_summary")
|
|
def test_planner_node(mock_get_summary, mock_get_llm, mock_state):
|
|
"""Test planner node with checklist generation."""
|
|
mock_get_summary.return_value = "Column: Name, Type: text"
|
|
|
|
mock_llm = MagicMock()
|
|
mock_get_llm.return_value = mock_llm
|
|
|
|
mock_response = ChecklistResponse(
|
|
goal="Get NJ results",
|
|
reflection="The user wants NJ results",
|
|
checklist=[
|
|
ChecklistTask(task="Query NJ data", worker="data_analyst")
|
|
]
|
|
)
|
|
mock_llm.with_structured_output.return_value.invoke.return_value = mock_response
|
|
|
|
result = planner_node(mock_state)
|
|
|
|
assert "checklist" in result
|
|
assert result["checklist"][0]["task"] == "Query NJ data"
|
|
assert result["current_step"] == 0
|
|
assert result["summary"] == "The user wants NJ results"
|
|
|
|
# Verify helper was called
|
|
mock_get_summary.assert_called_once()
|