"""Tests for src/planning_orchestrator.py — phase pipeline coordinator.""" from __future__ import annotations from unittest.mock import patch import pytest from src import planning_orchestrator, planning_session from src.planning_orchestrator import ( BASE_PHASES, DESIGN_PHASE, PlanningOrchestrator, _phases_for, has_ui_scope, ) @pytest.fixture def tmp_planning_state(tmp_path, monkeypatch): fake_sessions_dir = tmp_path / "sessions" fake_sessions_dir.mkdir() fake_state = fake_sessions_dir / "planning.json" monkeypatch.setattr(planning_session, "SESSIONS_DIR", fake_sessions_dir) monkeypatch.setattr(planning_session, "PLANNING_STATE_FILE", fake_state) yield fake_state @pytest.fixture def fake_workspace(tmp_path, monkeypatch): workspace = tmp_path / "workspace" workspace.mkdir() (workspace / "demo").mkdir() monkeypatch.setattr(planning_session, "WORKSPACE_ROOT", workspace) monkeypatch.setattr(planning_orchestrator, "WORKSPACE_ROOT", workspace) yield workspace # --------------------------------------------------------------------------- # has_ui_scope / _phases_for # --------------------------------------------------------------------------- class TestUiScopeHeuristic: @pytest.mark.parametrize("text,expected", [ ("redesign UI for the dashboard", True), ("add a button on settings page", True), ("frontend cleanup", True), ("Adaugă filtru genuri pe pagina de game-library", True), # ro ("schimbă culoarea butonului de submit", True), # ro ("refactor utility helpers", False), ("rewrite the database migration scripts", False), ("tweak the rate limiter", False), ]) def test_detects_ui_keywords(self, text, expected): assert has_ui_scope(text) is expected def test_phases_for_excludes_design_when_no_ui(self): phases = _phases_for("refactor utility") assert phases == BASE_PHASES assert DESIGN_PHASE not in phases def test_phases_for_appends_design_for_ui(self): phases = _phases_for("add login page") assert phases[-1] == DESIGN_PHASE assert phases[: len(BASE_PHASES)] == BASE_PHASES # --------------------------------------------------------------------------- # Orchestrator start / respond / advance / cancel — mock subprocess # --------------------------------------------------------------------------- def _fake_result(session_id="sess-1", text="hi"): return { "result": text, "session_id": session_id, "usage": {"input_tokens": 1000, "output_tokens": 200}, "total_cost_usd": 0.4, "subtype": "success", "is_error": False, } class TestOrchestratorStart: def test_start_persists_phases_planned( self, tmp_planning_state, fake_workspace ): with patch( "src.planning_session._run_claude", return_value=_fake_result(session_id="s-1"), ): sess, first = PlanningOrchestrator.start( "demo", "Add login button", "ch-1", adapter="discord" ) assert sess.session_id == "s-1" from src.planning_session import get_planning_state state = get_planning_state("discord", "ch-1") assert state["phase"] == BASE_PHASES[0] # /office-hours # UI scope → design phase included assert state["phases_planned"][-1] == DESIGN_PHASE assert state["phase_index"] == 0 assert state["phases_completed"] == [] def test_start_no_ui_scope_no_design_phase( self, tmp_planning_state, fake_workspace ): with patch( "src.planning_session._run_claude", return_value=_fake_result(session_id="s-1"), ): PlanningOrchestrator.start( "demo", "refactor utility helpers", "ch-1", adapter="discord" ) from src.planning_session import get_planning_state state = get_planning_state("discord", "ch-1") assert DESIGN_PHASE not in state["phases_planned"] class TestOrchestratorRespond: def test_respond_returns_text_and_phase_ready_marker( self, tmp_planning_state, fake_workspace ): with patch( "src.planning_session._run_claude", return_value=_fake_result(session_id="s-1", text="initial"), ): PlanningOrchestrator.start("demo", "Add login button", "ch-1", "discord") ready_text = "ok we are done. PHASE_STATUS: ready_to_advance" with patch( "src.planning_session._run_claude", return_value=_fake_result(session_id="s-1", text=ready_text), ): sess, response, ready = PlanningOrchestrator.respond( "discord", "ch-1", "user reply" ) assert response == ready_text assert ready is True assert sess is not None def test_respond_returns_none_when_no_state(self, tmp_planning_state): sess, text, ready = PlanningOrchestrator.respond( "discord", "ch-missing", "hi" ) assert sess is None assert "Nu există" in text assert ready is False class TestOrchestratorAdvance: def test_advance_starts_next_phase_fresh_subprocess( self, tmp_planning_state, fake_workspace ): # Phase 1 → office-hours with patch( "src.planning_session._run_claude", return_value=_fake_result(session_id="s-1"), ): PlanningOrchestrator.start( "demo", "Add login button", "ch-1", "discord" ) # Advance → /plan-ceo-review fresh subprocess with patch( "src.planning_session._run_claude", return_value=_fake_result(session_id="s-2", text="ceo phase started"), ) as mock_run: sess, text, completed = PlanningOrchestrator.advance( "discord", "ch-1" ) mock_run.assert_called_once() # Verify the new subprocess has /plan-ceo-review in prompt (NOT --resume) cmd = mock_run.call_args[0][0] assert "/plan-ceo-review" in cmd[2] assert "--resume" not in cmd assert sess.session_id == "s-2" assert sess.phase == "/plan-ceo-review" assert completed is False from src.planning_session import get_planning_state state = get_planning_state("discord", "ch-1") assert "/office-hours" in state["phases_completed"] assert state["phase_index"] == 1 def test_advance_writes_final_plan_when_pipeline_complete( self, tmp_planning_state, fake_workspace ): # Manually seed state at the last phase. from src.planning_session import _save_planning_state, _channel_key # Build phase plan with 2 phases for brevity (skip design for non-UI). state = { _channel_key("discord", "ch-1"): { "slug": "demo", "description": "refactor utility", "phase": "/plan-eng-review", "phases_planned": ["/office-hours", "/plan-ceo-review", "/plan-eng-review"], "phase_index": 2, "phases_completed": ["/office-hours", "/plan-ceo-review"], "session_id": "s-eng", "planning_session_id": "ps-uuid", "adapter": "discord", "channel_id": "ch-1", "started_at": "2026-04-26T20:00:00+00:00", "updated_at": "2026-04-26T20:30:00+00:00", "last_text_excerpt": "eng review done", "last_subtype": "success", } } _save_planning_state(state) # Advance with no more phases — should write final-plan stub, no claude call. with patch("src.planning_session._run_claude") as mock_run: sess, text, completed = PlanningOrchestrator.advance( "discord", "ch-1" ) mock_run.assert_not_called() assert completed is True plan_path = PlanningOrchestrator.final_plan_path("demo") assert plan_path.exists() content = plan_path.read_text(encoding="utf-8") assert "demo" in content assert "refactor utility" in content # All phases listed assert "/office-hours" in content assert "/plan-ceo-review" in content assert "/plan-eng-review" in content class TestOrchestratorCancel: def test_cancel_clears_state(self, tmp_planning_state, fake_workspace): with patch( "src.planning_session._run_claude", return_value=_fake_result(session_id="s-1"), ): PlanningOrchestrator.start("demo", "x", "ch-1", "discord") from src.planning_session import is_in_planning assert is_in_planning("discord", "ch-1") is True assert PlanningOrchestrator.cancel("discord", "ch-1") is True assert is_in_planning("discord", "ch-1") is False def test_cancel_returns_false_when_no_state(self, tmp_planning_state): assert PlanningOrchestrator.cancel("discord", "ch-x") is False