Spaces:
Runtime error
Runtime error
File size: 5,425 Bytes
9a49d21 b1b661d 9a49d21 a745005 9a49d21 c895509 9a49d21 b1b661d 9a49d21 | 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 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 | """Unit tests for LangGraph workflow state and nodes."""
import pytest
from unittest.mock import AsyncMock
from src.workflows.types import IntelligenceState
from src.workflows.market_analysis import MarketIntelligenceWorkflow
class TestIntelligenceState:
"""Test state schema validation."""
def test_state_has_required_fields(self):
"""Verify state schema has all required fields."""
required_fields = [
"research_type",
"company_name",
"industry",
"research_data",
"competitors",
"market_trends",
"raw_sources",
"swot",
"competitive_matrix",
"positioning",
"strategic_recommendations",
"executive_summary",
"full_report",
"report_metadata",
"current_agent",
"iteration",
"total_cost",
"total_tokens",
"errors",
"human_feedback",
"approved",
"revision_count",
]
state_annotations = IntelligenceState.__annotations__
for field in required_fields:
assert field in state_annotations, f"Missing required field: {field}"
class TestWorkflowInitialization:
"""Test workflow initialization."""
def test_workflow_initializes(self):
"""Test workflow can be initialized."""
workflow = MarketIntelligenceWorkflow()
assert workflow is not None
assert workflow.max_budget == 2.0
assert workflow.research_agent is not None
assert workflow.analysis_agent is not None
assert workflow.writer_agent is not None
def test_custom_budget(self):
"""Test workflow with custom budget."""
workflow = MarketIntelligenceWorkflow(max_budget=5.0)
assert workflow.max_budget == 5.0
class TestConditionalRouting:
"""Test workflow routing logic."""
def test_should_continue_to_analysis_success(self):
"""Test routing to analysis when research succeeds."""
workflow = MarketIntelligenceWorkflow()
state = {
"research_data": {"some": "data"},
"errors": [],
}
result = workflow._should_continue_to_analysis(state)
assert result == "analysis"
def test_should_continue_to_analysis_with_errors(self):
"""Test routing ends when research has errors."""
workflow = MarketIntelligenceWorkflow()
state = {
"research_data": {},
"errors": ["Some error"],
}
result = workflow._should_continue_to_analysis(state)
assert result == "end"
def test_should_continue_to_analysis_no_data(self):
"""Test routing ends when no research data."""
workflow = MarketIntelligenceWorkflow()
state = {
"errors": [],
}
result = workflow._should_continue_to_analysis(state)
assert result == "end"
def test_check_approval_approved(self):
"""Test approval check when approved."""
workflow = MarketIntelligenceWorkflow()
state = {
"approved": True,
"revision_count": 0,
}
result = workflow._check_approval(state)
assert result == "approved"
def test_check_approval_max_revisions(self):
"""Test approval check at max revisions."""
workflow = MarketIntelligenceWorkflow()
state = {
"approved": False,
"revision_count": 2,
}
result = workflow._check_approval(state)
assert result == "max_revisions"
def test_check_approval_revision_requested(self):
"""Test approval check with feedback."""
workflow = MarketIntelligenceWorkflow()
state = {
"approved": False,
"revision_count": 0,
"human_feedback": "Please add more details",
}
result = workflow._check_approval(state)
assert result == "revise"
@pytest.mark.asyncio
class TestWorkflowNodes:
"""Test individual workflow nodes."""
async def test_research_node_structure(self):
"""Test research node returns correct state structure."""
workflow = MarketIntelligenceWorkflow()
# Mock the research agent's run method
workflow.research_agent.run = AsyncMock(
return_value={
"company_name": "Test Company",
"company_overview": "Overview",
"competitors": "Competitors",
"market_trends": "Trends",
"raw_sources": [],
}
)
state = {
"company_name": "Test Company",
"industry": "Technology",
"iteration": 0,
}
result = await workflow._research_node(state)
assert "current_agent" in result
assert result["current_agent"] == "research"
assert "iteration" in result
async def test_human_review_node_auto_approves(self):
"""Test human review node auto-approves for now."""
workflow = MarketIntelligenceWorkflow()
state = {
"company_name": "Test Company",
}
result = await workflow._human_review_node(state)
assert result["current_agent"] == "human_review"
assert result["approved"] is True
assert result["human_feedback"] is None
|