Pathfinder MCP Server

"""Tests for implementation tools.""" import pytest from pathfinder_mcp.state import Phase from pathfinder_mcp.tools.implement import _extract_phases, implement_phase class TestExtractPhases: """Tests for phase extraction from plan.""" def test_extracts_phases_from_plan(self, sample_plan_content: str) -> None: """Extracts phase sections from plan content.""" phases = _extract_phases(sample_plan_content) assert len(phases) == 2 assert phases[0]["number"] == 1 assert phases[1]["number"] == 2 def test_extracts_tasks(self, sample_plan_content: str) -> None: """Extracts tasks from phase sections.""" phases = _extract_phases(sample_plan_content) # Phase 1 should have tasks assert len(phases[0]["tasks"]) > 0 def test_handles_plan_without_phases(self) -> None: """Returns empty list for plan without phases.""" content = "# Plan\n\nNo phases here." phases = _extract_phases(content) assert phases == [] class TestImplementPhase: """Tests for implement_phase tool.""" @pytest.mark.asyncio async def test_transitions_to_implement( self, session_manager, artifact_writer, context_monitor, active_sessions, sample_session_id, plan_state, sample_plan_content, ) -> None: """implement_phase transitions from PLAN to IMPLEMENT.""" session_manager.create_session(sample_session_id) active_sessions[sample_session_id] = plan_state artifact_writer.write_plan(sample_session_id, sample_plan_content) result = await implement_phase( session_id=sample_session_id, phase_number=None, ctx=None, session_manager=session_manager, artifact_writer=artifact_writer, context_monitor=context_monitor, sessions=active_sessions, ) assert result["phase"] == "implement" assert active_sessions[sample_session_id].current_phase == Phase.IMPLEMENT @pytest.mark.asyncio async def test_returns_phase_tasks( self, session_manager, artifact_writer, context_monitor, active_sessions, sample_session_id, plan_state, sample_plan_content, ) -> None: """implement_phase returns tasks for the phase.""" session_manager.create_session(sample_session_id) active_sessions[sample_session_id] = plan_state artifact_writer.write_plan(sample_session_id, sample_plan_content) result = await implement_phase( session_id=sample_session_id, phase_number=1, ctx=None, session_manager=session_manager, artifact_writer=artifact_writer, context_monitor=context_monitor, sessions=active_sessions, ) assert "executing_phase" in result assert result["executing_phase"]["number"] == 1 assert "tasks" in result["executing_phase"] @pytest.mark.asyncio async def test_updates_progress_artifact( self, session_manager, artifact_writer, context_monitor, active_sessions, sample_session_id, plan_state, sample_plan_content, ) -> None: """implement_phase creates progress.md entry.""" session_manager.create_session(sample_session_id) active_sessions[sample_session_id] = plan_state artifact_writer.write_plan(sample_session_id, sample_plan_content) await implement_phase( session_id=sample_session_id, phase_number=1, ctx=None, session_manager=session_manager, artifact_writer=artifact_writer, context_monitor=context_monitor, sessions=active_sessions, ) progress = artifact_writer.read_artifact(sample_session_id, "progress.md") assert progress is not None assert "Phase 1" in progress @pytest.mark.asyncio async def test_fails_from_research_phase( self, session_manager, artifact_writer, context_monitor, active_sessions, sample_session_id, research_state, ) -> None: """implement_phase fails when in RESEARCH phase.""" session_manager.create_session(sample_session_id) active_sessions[sample_session_id] = research_state result = await implement_phase( session_id=sample_session_id, phase_number=None, ctx=None, session_manager=session_manager, artifact_writer=artifact_writer, context_monitor=context_monitor, sessions=active_sessions, ) assert result["code"] == "INVALID_PHASE" @pytest.mark.asyncio async def test_fails_without_plan( self, session_manager, artifact_writer, context_monitor, active_sessions, sample_session_id, plan_state, ) -> None: """implement_phase fails without plan.md.""" session_manager.create_session(sample_session_id) active_sessions[sample_session_id] = plan_state # Don't create plan result = await implement_phase( session_id=sample_session_id, phase_number=None, ctx=None, session_manager=session_manager, artifact_writer=artifact_writer, context_monitor=context_monitor, sessions=active_sessions, ) assert result["code"] == "MISSING_PLAN" @pytest.mark.asyncio async def test_selects_specific_phase( self, session_manager, artifact_writer, context_monitor, active_sessions, sample_session_id, implement_state, sample_plan_content, ) -> None: """implement_phase can select specific phase number.""" session_manager.create_session(sample_session_id) active_sessions[sample_session_id] = implement_state artifact_writer.write_plan(sample_session_id, sample_plan_content) result = await implement_phase( session_id=sample_session_id, phase_number=2, ctx=None, session_manager=session_manager, artifact_writer=artifact_writer, context_monitor=context_monitor, sessions=active_sessions, ) assert result["executing_phase"]["number"] == 2 @pytest.mark.asyncio async def test_returns_error_for_invalid_phase_number( self, session_manager, artifact_writer, context_monitor, active_sessions, sample_session_id, implement_state, sample_plan_content, ) -> None: """implement_phase returns error for non-existent phase.""" session_manager.create_session(sample_session_id) active_sessions[sample_session_id] = implement_state artifact_writer.write_plan(sample_session_id, sample_plan_content) result = await implement_phase( session_id=sample_session_id, phase_number=99, ctx=None, session_manager=session_manager, artifact_writer=artifact_writer, context_monitor=context_monitor, sessions=active_sessions, ) assert result["code"] == "PHASE_NOT_FOUND" assert "available_phases" in result