""" Tests for the dynamic graph builder (Phase 3). Verifies that build_graph_from_blueprint correctly creates LangGraph graphs from JSON blueprints matching the frontend's CouncilBlueprint format. All LLM calls are mocked. """ import sys import os sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) import pytest from unittest.mock import patch, MagicMock from services.dynamic_graph_builder import ( build_graph_from_blueprint, _make_agent_node, _make_critic_node, _make_conditional_router, _is_critic_like, _get_llm, ) from services.graph_builder import create_initial_state from state import CouncilState, APPROVAL_THRESHOLD, MAX_ITERATIONS # --------------------------------------------------------------------------- # Sample blueprints for testing # --------------------------------------------------------------------------- SIMPLE_LINEAR_BLUEPRINT = { "version": 1, "name": "Simple Linear", "nodes": [ { "id": "node-1", "label": "Writer", "systemPrompt": "You are a professional writer.", "model": "claude-3-5-sonnet", "tools": {"webSearch": False, "pdfReader": False}, "position": {"x": 0, "y": 0}, }, { "id": "node-2", "label": "Editor", "systemPrompt": "You are a professional editor. Polish the text.", "model": "claude-3-5-sonnet", "tools": {"webSearch": False, "pdfReader": False}, "position": {"x": 300, "y": 0}, }, ], "edges": [ {"id": "edge-1", "source": "node-1", "target": "node-2", "type": "linear"}, ], } CYCLIC_BLUEPRINT = { "version": 1, "name": "Cyclic Council", "nodes": [ { "id": "master", "label": "Master Agent", "systemPrompt": "You are the master writer.", "model": "claude-3-5-sonnet", "tools": {"webSearch": False, "pdfReader": False}, "position": {"x": 0, "y": 0}, }, { "id": "critic", "label": "Critic Agent", "systemPrompt": "You are the critic. Evaluate and score the draft.", "model": "claude-3-5-sonnet", "tools": {"webSearch": False, "pdfReader": False}, "position": {"x": 300, "y": 0}, }, { "id": "writer", "label": "Final Writer", "systemPrompt": "You polish approved drafts.", "model": "claude-3-5-sonnet", "tools": {"webSearch": False, "pdfReader": False}, "position": {"x": 600, "y": 0}, }, ], "edges": [ {"id": "e1", "source": "master", "target": "critic", "type": "linear"}, { "id": "e2", "source": "critic", "target": "master", "type": "conditional", "condition": "rework", }, { "id": "e3", "source": "critic", "target": "writer", "type": "conditional", "condition": "approve", }, ], } # --------------------------------------------------------------------------- # Test: critic detection heuristic # --------------------------------------------------------------------------- class TestCriticDetection: def test_detects_critic_keyword(self): assert _is_critic_like("You are the critic. Evaluate drafts.") is True def test_detects_evaluate_keyword(self): assert _is_critic_like("Your role is to evaluate and score.") is True def test_detects_review_keyword(self): assert _is_critic_like("Review the document for quality.") is True def test_no_match_for_writer(self): assert _is_critic_like("You are a professional writer.") is False def test_case_insensitive(self): assert _is_critic_like("You are the CRITIC agent.") is True # --------------------------------------------------------------------------- # Test: conditional routing # --------------------------------------------------------------------------- class TestConditionalRouter: def test_routes_to_correct_target(self): edges = [ {"target": "node-a", "condition": "rework"}, {"target": "node-b", "condition": "approve"}, ] router = _make_conditional_router("source", edges, None) state = create_initial_state("topic", "run-1") state["route_decision"] = "approve" assert router(state) == "node-b" def test_routes_rework(self): edges = [ {"target": "node-a", "condition": "rework"}, {"target": "node-b", "condition": "approve"}, ] router = _make_conditional_router("source", edges, None) state = create_initial_state("topic", "run-1") state["route_decision"] = "rework" assert router(state) == "node-a" def test_unknown_decision_uses_linear_fallback(self): edges = [ {"target": "node-a", "condition": "rework"}, ] router = _make_conditional_router("source", edges, "fallback-node") state = create_initial_state("topic", "run-1") state["route_decision"] = "unknown" assert router(state) == "fallback-node" def test_unknown_decision_uses_first_conditional_as_fallback(self): edges = [ {"target": "node-a", "condition": "rework"}, {"target": "node-b", "condition": "approve"}, ] router = _make_conditional_router("source", edges, None) state = create_initial_state("topic", "run-1") state["route_decision"] = "unknown" assert router(state) == "node-a" # --------------------------------------------------------------------------- # Test: agent node factory # --------------------------------------------------------------------------- class TestAgentNodeFactory: def test_agent_node_returns_draft(self): mock_response = MagicMock() mock_response.content = "Generated content about AI." with patch("services.dynamic_graph_builder.ChatAnthropic") as MockLLM: MockLLM.return_value.invoke.return_value = mock_response node_fn = _make_agent_node("node-1", "Writer", "You write.", "claude-3-5-sonnet") state = create_initial_state("AI basics", "run-1") result = node_fn(state) assert result["current_draft"] == "Generated content about AI." assert result["active_node"] == "node-1" assert result["iteration_count"] == 1 def test_agent_node_with_existing_draft_and_feedback(self): mock_response = MagicMock() mock_response.content = "Improved draft." with patch("services.dynamic_graph_builder.ChatAnthropic") as MockLLM: MockLLM.return_value.invoke.return_value = mock_response node_fn = _make_agent_node("node-1", "Writer", "You write.", "claude-3-5-sonnet") state = create_initial_state("AI", "run-1") state["current_draft"] = "First draft" state["feedback_history"] = ["Needs more detail"] state["iteration_count"] = 1 result = node_fn(state) assert result["current_draft"] == "Improved draft." assert result["iteration_count"] == 2 # --------------------------------------------------------------------------- # Test: critic node factory # --------------------------------------------------------------------------- class TestCriticNodeFactory: def test_critic_node_approves_high_score(self): mock_response = MagicMock() mock_response.content = "SCORE: 9\nVERDICT: approve\nFEEDBACK:\nExcellent work." with patch("services.dynamic_graph_builder.ChatAnthropic") as MockLLM: MockLLM.return_value.invoke.return_value = mock_response node_fn = _make_critic_node("critic-1", "Critic", "You evaluate.", "claude-3-5-sonnet") state = create_initial_state("Topic", "run-1") state["current_draft"] = "A great draft" result = node_fn(state) assert result["route_decision"] == "approve" assert result["critic_score"] == 9.0 def test_critic_node_reworks_low_score(self): mock_response = MagicMock() mock_response.content = "SCORE: 4\nVERDICT: rework\nFEEDBACK:\nNeeds more structure." with patch("services.dynamic_graph_builder.ChatAnthropic") as MockLLM: MockLLM.return_value.invoke.return_value = mock_response node_fn = _make_critic_node("critic-1", "Critic", "You evaluate.", "claude-3-5-sonnet") state = create_initial_state("Topic", "run-1") state["current_draft"] = "Draft" result = node_fn(state) assert result["route_decision"] == "rework" assert result["critic_score"] == 4.0 assert len(result["feedback_history"]) == 1 assert "structure" in result["feedback_history"][0] def test_critic_safety_valve_at_max_iterations(self): node_fn = _make_critic_node("critic-1", "Critic", "Evaluate.", "claude-3-5-sonnet") state = create_initial_state("Topic", "run-1") state["current_draft"] = "Draft" state["iteration_count"] = MAX_ITERATIONS result = node_fn(state) assert result["route_decision"] == "approve" assert result["critic_score"] == APPROVAL_THRESHOLD # --------------------------------------------------------------------------- # Test: build_graph_from_blueprint # --------------------------------------------------------------------------- class TestBuildGraphFromBlueprint: def test_rejects_empty_blueprint(self): with pytest.raises(ValueError, match="no nodes"): build_graph_from_blueprint({"version": 1, "name": "Empty", "nodes": [], "edges": []}) def test_builds_linear_graph(self): """A simple linear blueprint should compile without error.""" graph = build_graph_from_blueprint(SIMPLE_LINEAR_BLUEPRINT) assert graph is not None def test_builds_cyclic_graph(self): """A cyclic blueprint with conditional edges should compile.""" graph = build_graph_from_blueprint(CYCLIC_BLUEPRINT) assert graph is not None def test_entry_point_is_node_with_no_incoming(self): """The entry point should be the node that has no incoming edges.""" # In CYCLIC_BLUEPRINT, 'master' has no incoming edges except from critic (conditional rework), # but critic->master is an edge so master IS a target. The first node without incoming = master # Actually master IS a target of the rework edge. Let's verify with simple linear. graph = build_graph_from_blueprint(SIMPLE_LINEAR_BLUEPRINT) assert graph is not None # node-1 has no incoming, so it's the entry def test_single_node_blueprint(self): """A single node with no edges should work (trivial graph).""" bp = { "version": 1, "name": "Single", "nodes": [ { "id": "only-node", "label": "Solo Agent", "systemPrompt": "You work alone.", "model": "claude-3-5-sonnet", "tools": {"webSearch": False, "pdfReader": False}, "position": {"x": 0, "y": 0}, } ], "edges": [], } graph = build_graph_from_blueprint(bp) assert graph is not None # --------------------------------------------------------------------------- # Test: model lookup # --------------------------------------------------------------------------- class TestModelLookup: def test_unknown_model_raises(self): with pytest.raises(ValueError, match="Unknown model"): _get_llm("nonexistent-model") def test_claude_model_creates_instance(self): with patch("services.dynamic_graph_builder.ChatAnthropic") as MockLLM: llm = _get_llm("claude-3-5-sonnet") MockLLM.assert_called_once() def test_gpt4o_model_creates_instance(self): with patch("services.dynamic_graph_builder.ChatOpenAI") as MockLLM: llm = _get_llm("gpt-4o") MockLLM.assert_called_once()