diff --git a/src/aipass/flow/tests/test_push_branch_dashboard.py b/src/aipass/flow/tests/test_push_branch_dashboard.py new file mode 100644 index 00000000..b4a01df8 --- /dev/null +++ b/src/aipass/flow/tests/test_push_branch_dashboard.py @@ -0,0 +1,729 @@ +# =================== AIPass ==================== +# Name: test_push_branch_dashboard.py +# Description: Tests for push_branch_dashboard handler — branch dashboard push +# Version: 1.0.0 +# Created: 2026-04-26 +# Modified: 2026-04-26 +# ============================================= + +"""Tests for push_branch_dashboard handler — branch dashboard push.""" + +import json +from datetime import datetime, timezone, timedelta +from unittest.mock import patch + +_MOD = "aipass.flow.apps.handlers.dashboard.push_branch_dashboard" + + +# ─── Import helpers ─────────────────────────────────────── + + +def _import_mod(): + import aipass.flow.apps.handlers.dashboard.push_branch_dashboard as mod + + return mod + + +# ═══════════════════════════════════════════════════════════ +# 1. _write_dashboard_section +# ═══════════════════════════════════════════════════════════ + + +class TestWriteDashboardSection: + """Tests for _write_dashboard_section.""" + + def test_creates_dashboard_when_not_exists(self, tmp_path): + """Creates DASHBOARD.local.json when it does not exist.""" + mod = _import_mod() + section_data = {"managed_by": "flow", "active_plans": [], "active_count": 0} + with patch.object(mod, "DASHBOARD_TEMPLATE_FILE", tmp_path / "nonexistent_template.json"): + result = mod._write_dashboard_section(tmp_path, "flow", section_data) + assert result is True + dashboard_path = tmp_path / "DASHBOARD.local.json" + assert dashboard_path.exists() + dashboard = json.loads(dashboard_path.read_text(encoding="utf-8")) + assert "flow" in dashboard["sections"] + assert dashboard["sections"]["flow"]["managed_by"] == "flow" + + def test_updates_existing_dashboard(self, tmp_path): + """Updates an existing DASHBOARD.local.json with new section data.""" + mod = _import_mod() + existing = { + "branch": "TEST", + "last_updated": "", + "quick_status": {}, + "sections": { + "ai_mail": {"managed_by": "ai_mail", "new": 0}, + }, + } + dashboard_path = tmp_path / "DASHBOARD.local.json" + dashboard_path.write_text(json.dumps(existing), encoding="utf-8") + + section_data = {"managed_by": "flow", "active_plans": [], "active_count": 0} + result = mod._write_dashboard_section(tmp_path, "flow", section_data) + + assert result is True + updated = json.loads(dashboard_path.read_text(encoding="utf-8")) + assert "flow" in updated["sections"] + assert "ai_mail" in updated["sections"] + + def test_corrupt_json_creates_fresh(self, tmp_path): + """Corrupt JSON in dashboard file triggers creation of fresh dashboard.""" + mod = _import_mod() + dashboard_path = tmp_path / "DASHBOARD.local.json" + dashboard_path.write_text("{not valid json!!!", encoding="utf-8") + + section_data = {"managed_by": "flow", "active_count": 0} + with patch.object(mod, "DASHBOARD_TEMPLATE_FILE", tmp_path / "nonexistent_template.json"): + result = mod._write_dashboard_section(tmp_path, "flow", section_data) + + assert result is True + updated = json.loads(dashboard_path.read_text(encoding="utf-8")) + assert "sections" in updated + assert "flow" in updated["sections"] + + def test_empty_file_creates_fresh(self, tmp_path): + """Empty dashboard file triggers creation of fresh dashboard.""" + mod = _import_mod() + dashboard_path = tmp_path / "DASHBOARD.local.json" + dashboard_path.write_text("", encoding="utf-8") + + section_data = {"managed_by": "flow", "active_count": 0} + with patch.object(mod, "DASHBOARD_TEMPLATE_FILE", tmp_path / "nonexistent_template.json"): + result = mod._write_dashboard_section(tmp_path, "flow", section_data) + + assert result is True + updated = json.loads(dashboard_path.read_text(encoding="utf-8")) + assert "sections" in updated + + def test_recalculates_quick_status(self, tmp_path): + """Dashboard quick_status is recalculated after section write.""" + mod = _import_mod() + existing = { + "branch": "TEST", + "last_updated": "", + "quick_status": {}, + "sections": { + "ai_mail": {"managed_by": "ai_mail", "new": 3}, + }, + } + dashboard_path = tmp_path / "DASHBOARD.local.json" + dashboard_path.write_text(json.dumps(existing), encoding="utf-8") + + section_data = {"managed_by": "flow", "active_count": 2} + result = mod._write_dashboard_section(tmp_path, "flow", section_data) + + assert result is True + updated = json.loads(dashboard_path.read_text(encoding="utf-8")) + assert updated["quick_status"]["action_required"] is True + assert updated["quick_status"]["new_mail"] == 3 + assert updated["quick_status"]["active_plans"] == 2 + + def test_returns_false_on_exception(self, tmp_path): + """Returns False when an exception occurs during write.""" + mod = _import_mod() + section_data = {"managed_by": "flow"} + # Patch Path to simulate write failure + with patch.object(mod, "_create_fresh_dashboard", side_effect=RuntimeError("boom")): + result = mod._write_dashboard_section(tmp_path, "flow", section_data) + assert result is False + + def test_adds_last_updated_to_section(self, tmp_path): + """Section data gets a last_updated timestamp injected.""" + mod = _import_mod() + dashboard_path = tmp_path / "DASHBOARD.local.json" + dashboard_path.write_text(json.dumps({"sections": {}}), encoding="utf-8") + + section_data = {"managed_by": "flow", "active_count": 0} + mod._write_dashboard_section(tmp_path, "flow", section_data) + + updated = json.loads(dashboard_path.read_text(encoding="utf-8")) + assert "last_updated" in updated["sections"]["flow"] + assert "last_updated" in updated + + +# ═══════════════════════════════════════════════════════════ +# 2. _create_fresh_dashboard +# ═══════════════════════════════════════════════════════════ + + +class TestCreateFreshDashboard: + """Tests for _create_fresh_dashboard.""" + + def test_uses_template_when_exists(self, tmp_path): + """Loads from DASHBOARD_TEMPLATE_FILE when it exists.""" + mod = _import_mod() + template = { + "branch": "{{BRANCHNAME}}", + "sections": {}, + "last_updated": "", + } + template_file = tmp_path / "DASHBOARD.template.json" + template_file.write_text(json.dumps(template), encoding="utf-8") + + branch_path = tmp_path / "my_branch" + branch_path.mkdir() + + with patch.object(mod, "DASHBOARD_TEMPLATE_FILE", template_file): + result = mod._create_fresh_dashboard(branch_path) + + assert result["branch"] == "MY_BRANCH" + assert "last_updated" in result + + def test_fallback_when_no_template(self, tmp_path): + """Falls back to hardcoded defaults when template file does not exist.""" + mod = _import_mod() + branch_path = tmp_path / "test_branch" + branch_path.mkdir() + + with patch.object(mod, "DASHBOARD_TEMPLATE_FILE", tmp_path / "nonexistent.json"): + result = mod._create_fresh_dashboard(branch_path) + + assert result["branch"] == "TEST_BRANCH" + assert result["_warning"] == "AUTO-GENERATED FILE - DO NOT MANUALLY EDIT." + assert "ai_mail" in result["sections"] + assert "flow" in result["sections"] + assert "memory" in result["sections"] + assert "devpulse" in result["sections"] + assert "commons_activity" in result["sections"] + + def test_fallback_on_corrupt_template(self, tmp_path): + """Falls back to defaults when template file contains invalid JSON.""" + mod = _import_mod() + template_file = tmp_path / "DASHBOARD.template.json" + template_file.write_text("not json at all", encoding="utf-8") + + branch_path = tmp_path / "fallback_branch" + branch_path.mkdir() + + with patch.object(mod, "DASHBOARD_TEMPLATE_FILE", template_file): + result = mod._create_fresh_dashboard(branch_path) + + assert result["branch"] == "FALLBACK_BRANCH" + assert "sections" in result + + +# ═══════════════════════════════════════════════════════════ +# 3. _calculate_quick_status +# ═══════════════════════════════════════════════════════════ + + +class TestCalculateQuickStatus: + """Tests for _calculate_quick_status.""" + + def test_all_clear_when_nothing(self): + """Returns 'All clear' summary when all counts are zero.""" + mod = _import_mod() + sections = { + "ai_mail": {"new": 0, "opened": 0}, + "flow": {"active_count": 0}, + "commons_activity": {"mentions": 0}, + } + result = mod._calculate_quick_status(sections) + assert result["action_required"] is False + assert result["summary"] == "All clear" + assert result["new_mail"] == 0 + + def test_action_required_with_new_mail(self): + """Sets action_required True when there is new mail.""" + mod = _import_mod() + sections = { + "ai_mail": {"new": 5, "opened": 1}, + "flow": {"active_count": 0}, + "commons_activity": {"mentions": 0}, + } + result = mod._calculate_quick_status(sections) + assert result["action_required"] is True + assert "5 new emails" in result["summary"] + assert "1 opened" in result["summary"] + + def test_active_plans_as_list(self): + """Handles active_count being a list by taking len().""" + mod = _import_mod() + sections = { + "ai_mail": {"new": 0}, + "flow": {"active_count": ["plan1", "plan2", "plan3"]}, + "commons_activity": {"mentions": 0}, + } + result = mod._calculate_quick_status(sections) + assert result["active_plans"] == 3 + assert result["action_required"] is True + assert "3 active plans" in result["summary"] + + def test_active_plans_as_int(self): + """Handles active_count as an integer directly.""" + mod = _import_mod() + sections = { + "ai_mail": {"new": 0}, + "flow": {"active_count": 2}, + "commons_activity": {"mentions": 0}, + } + result = mod._calculate_quick_status(sections) + assert result["active_plans"] == 2 + + def test_mentions_trigger_action_required(self): + """Commons mentions trigger action_required.""" + mod = _import_mod() + sections = { + "ai_mail": {"new": 0}, + "flow": {"active_count": 0}, + "commons_activity": {"mentions": 4}, + } + result = mod._calculate_quick_status(sections) + assert result["action_required"] is True + assert "4 mentions" in result["summary"] + + def test_empty_sections(self): + """Handles completely empty sections dict.""" + mod = _import_mod() + result = mod._calculate_quick_status({}) + assert result["action_required"] is False + assert result["summary"] == "All clear" + assert result["new_mail"] == 0 + assert result["active_plans"] == 0 + assert result["commons_mentions"] == 0 + + def test_uses_unread_fallback(self): + """Falls back to 'unread' key when 'new' is missing from ai_mail.""" + mod = _import_mod() + sections = { + "ai_mail": {"unread": 7}, + "flow": {"active_count": 0}, + "commons_activity": {"mentions": 0}, + } + result = mod._calculate_quick_status(sections) + assert result["new_mail"] == 7 + assert result["action_required"] is True + + +# ═══════════════════════════════════════════════════════════ +# 4. _get_all_registry_files +# ═══════════════════════════════════════════════════════════ + + +class TestGetAllRegistryFiles: + """Tests for _get_all_registry_files.""" + + def test_reads_template_registry(self, tmp_path): + """Reads per-type registry filenames from template_registry.json.""" + mod = _import_mod() + template_reg = { + "types": { + "flow_plans": {"prefix": "FPLAN"}, + "dev_plans": {"prefix": "DPLAN"}, + } + } + reg_file = tmp_path / "template_registry.json" + reg_file.write_text(json.dumps(template_reg), encoding="utf-8") + + with patch.object(mod, "FLOW_JSON_DIR", tmp_path): + result = mod._get_all_registry_files() + + assert "fplan_registry.json" in result + assert "dplan_registry.json" in result + assert len(result) == 2 + + def test_falls_back_on_missing_template_registry(self, tmp_path): + """Falls back to REGISTRY_FILE.name when template_registry.json is missing.""" + mod = _import_mod() + with patch.object(mod, "FLOW_JSON_DIR", tmp_path): + result = mod._get_all_registry_files() + assert result == [mod.REGISTRY_FILE.name] + + def test_deduplicates_prefixes(self, tmp_path): + """Does not duplicate registry filenames for repeated prefixes.""" + mod = _import_mod() + template_reg = { + "types": { + "flow_plans": {"prefix": "FPLAN"}, + "flow_plans_v2": {"prefix": "FPLAN"}, + } + } + reg_file = tmp_path / "template_registry.json" + reg_file.write_text(json.dumps(template_reg), encoding="utf-8") + + with patch.object(mod, "FLOW_JSON_DIR", tmp_path): + result = mod._get_all_registry_files() + + assert result.count("fplan_registry.json") == 1 + + +# ═══════════════════════════════════════════════════════════ +# 5. _load_registry +# ═══════════════════════════════════════════════════════════ + + +class TestLoadRegistry: + """Tests for _load_registry.""" + + def test_merges_multiple_registries(self, tmp_path): + """Merges plans from multiple registry files.""" + mod = _import_mod() + fplan_reg = {"plans": {"1": {"subject": "fplan one"}}, "next_number": 5} + dplan_reg = {"plans": {"2": {"subject": "dplan one"}}, "next_number": 10} + (tmp_path / "fplan_registry.json").write_text(json.dumps(fplan_reg), encoding="utf-8") + (tmp_path / "dplan_registry.json").write_text(json.dumps(dplan_reg), encoding="utf-8") + + with ( + patch.object(mod, "FLOW_JSON_DIR", tmp_path), + patch.object(mod, "_get_all_registry_files", return_value=["fplan_registry.json", "dplan_registry.json"]), + ): + result = mod._load_registry() + + assert "1" in result["plans"] + assert "2" in result["plans"] + assert result["next_number"] == 10 + + def test_handles_missing_registry(self, tmp_path): + """Gracefully handles a missing registry file.""" + mod = _import_mod() + with ( + patch.object(mod, "FLOW_JSON_DIR", tmp_path), + patch.object(mod, "_get_all_registry_files", return_value=["nonexistent_registry.json"]), + ): + result = mod._load_registry() + assert result["plans"] == {} + assert result["next_number"] == 1 + + def test_keeps_highest_next_number(self, tmp_path): + """Keeps the highest next_number across registries.""" + mod = _import_mod() + reg_a = {"plans": {}, "next_number": 3} + reg_b = {"plans": {}, "next_number": 50} + reg_c = {"plans": {}, "next_number": 20} + (tmp_path / "a_registry.json").write_text(json.dumps(reg_a), encoding="utf-8") + (tmp_path / "b_registry.json").write_text(json.dumps(reg_b), encoding="utf-8") + (tmp_path / "c_registry.json").write_text(json.dumps(reg_c), encoding="utf-8") + + with ( + patch.object(mod, "FLOW_JSON_DIR", tmp_path), + patch.object( + mod, + "_get_all_registry_files", + return_value=["a_registry.json", "b_registry.json", "c_registry.json"], + ), + ): + result = mod._load_registry() + assert result["next_number"] == 50 + + def test_handles_corrupt_registry_gracefully(self, tmp_path): + """Skips a corrupt registry file and continues with others.""" + mod = _import_mod() + (tmp_path / "bad_registry.json").write_text("not json!", encoding="utf-8") + good_reg = {"plans": {"1": {"subject": "good"}}, "next_number": 5} + (tmp_path / "good_registry.json").write_text(json.dumps(good_reg), encoding="utf-8") + + with ( + patch.object(mod, "FLOW_JSON_DIR", tmp_path), + patch.object( + mod, + "_get_all_registry_files", + return_value=["bad_registry.json", "good_registry.json"], + ), + ): + result = mod._load_registry() + + assert "1" in result["plans"] + assert result["next_number"] == 5 + + +# ═══════════════════════════════════════════════════════════ +# 6. _filter_branch_plans +# ═══════════════════════════════════════════════════════════ + + +class TestFilterBranchPlans: + """Tests for _filter_branch_plans.""" + + def test_filters_active_plans_for_branch(self, tmp_path): + """Returns active plans matching the branch path.""" + mod = _import_mod() + registry = { + "plans": { + "1": { + "subject": "Plan A", + "status": "open", + "created": "2026-04-20", + "file_path": str(tmp_path / "FPLAN-0001_plan_a.md"), + "location": str(tmp_path), + }, + "2": { + "subject": "Plan B", + "status": "open", + "created": "2026-04-22", + "file_path": str(tmp_path / "FPLAN-0002_plan_b.md"), + "location": str(tmp_path), + }, + } + } + active, closed, total = mod._filter_branch_plans(registry, tmp_path) + assert len(active) == 2 + assert len(closed) == 0 + assert total == 2 + # Newest first + assert active[0]["id"] == "FPLAN-0002" + + def test_no_plans_for_branch(self, tmp_path): + """Returns empty lists when no plans match the branch.""" + mod = _import_mod() + other_path = tmp_path / "other_branch" + other_path.mkdir() + registry = { + "plans": { + "1": { + "subject": "Elsewhere", + "status": "open", + "created": "2026-04-20", + "file_path": "/somewhere/else/FPLAN-0001.md", + "location": str(other_path), + }, + } + } + active, closed, total = mod._filter_branch_plans(registry, tmp_path) + assert active == [] + assert closed == [] + assert total == 0 + + def test_recently_closed_within_7_days(self, tmp_path): + """Includes closed plans within the 7-day window.""" + mod = _import_mod() + recent_ts = (datetime.now(timezone.utc) - timedelta(days=2)).isoformat() + registry = { + "plans": { + "1": { + "subject": "Recently closed", + "status": "closed", + "created": "2026-04-18", + "closed": recent_ts, + "file_path": str(tmp_path / "FPLAN-0001_recent.md"), + "location": str(tmp_path), + }, + } + } + active, closed, total = mod._filter_branch_plans(registry, tmp_path) + assert len(closed) == 1 + assert closed[0]["id"] == "FPLAN-0001" + assert total == 1 + + def test_excludes_old_closed_plans(self, tmp_path): + """Excludes closed plans older than 7 days.""" + mod = _import_mod() + old_ts = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat() + registry = { + "plans": { + "1": { + "subject": "Old closed", + "status": "closed", + "created": "2026-03-01", + "closed": old_ts, + "file_path": str(tmp_path / "FPLAN-0001_old.md"), + "location": str(tmp_path), + }, + } + } + active, closed, total = mod._filter_branch_plans(registry, tmp_path) + assert len(closed) == 0 + assert total == 1 + + def test_recently_closed_capped_at_5(self, tmp_path): + """Recently closed list is limited to 5 entries.""" + mod = _import_mod() + plans = {} + for i in range(1, 9): + ts = (datetime.now(timezone.utc) - timedelta(hours=i)).isoformat() + plans[str(i)] = { + "subject": f"Closed plan {i}", + "status": "closed", + "created": "2026-04-20", + "closed": ts, + "file_path": str(tmp_path / f"FPLAN-{str(i).zfill(4)}_closed_{i}.md"), + "location": str(tmp_path), + } + registry = {"plans": plans} + active, closed, total = mod._filter_branch_plans(registry, tmp_path) + assert len(closed) == 5 + assert total == 8 + + def test_unparseable_closed_timestamp_included_anyway(self, tmp_path): + """Plans with unparseable closed timestamps are included anyway.""" + mod = _import_mod() + registry = { + "plans": { + "1": { + "subject": "Bad timestamp", + "status": "closed", + "created": "2026-04-20", + "closed": "not-a-date", + "file_path": str(tmp_path / "FPLAN-0001_bad_ts.md"), + "location": str(tmp_path), + }, + } + } + active, closed, total = mod._filter_branch_plans(registry, tmp_path) + assert len(closed) == 1 + assert closed[0]["closed"] == "not-a-date" + + def test_sorts_active_newest_first(self, tmp_path): + """Active plans are sorted by created date, newest first.""" + mod = _import_mod() + registry = { + "plans": { + "1": { + "subject": "Oldest", + "status": "open", + "created": "2026-04-01", + "file_path": str(tmp_path / "FPLAN-0001_oldest.md"), + "location": str(tmp_path), + }, + "2": { + "subject": "Newest", + "status": "open", + "created": "2026-04-25", + "file_path": str(tmp_path / "FPLAN-0002_newest.md"), + "location": str(tmp_path), + }, + "3": { + "subject": "Middle", + "status": "open", + "created": "2026-04-15", + "file_path": str(tmp_path / "FPLAN-0003_middle.md"), + "location": str(tmp_path), + }, + } + } + active, _, _ = mod._filter_branch_plans(registry, tmp_path) + assert active[0]["subject"] == "Newest" + assert active[1]["subject"] == "Middle" + assert active[2]["subject"] == "Oldest" + + def test_extracts_plan_prefix_from_filepath(self, tmp_path): + """Extracts correct plan prefix (DPLAN, TDPLAN, etc.) from file_path.""" + mod = _import_mod() + registry = { + "plans": { + "42": { + "subject": "Dev plan", + "status": "open", + "created": "2026-04-20", + "file_path": str(tmp_path / "DPLAN-0042_dev_plan.md"), + "location": str(tmp_path), + }, + } + } + active, _, _ = mod._filter_branch_plans(registry, tmp_path) + assert active[0]["id"] == "DPLAN-0042" + + +# ═══════════════════════════════════════════════════════════ +# 7. _build_section_data +# ═══════════════════════════════════════════════════════════ + + +class TestBuildSectionData: + """Tests for _build_section_data.""" + + def test_builds_correct_structure(self): + """Returns section dict with all expected keys.""" + mod = _import_mod() + active = [{"id": "FPLAN-0001", "subject": "Test"}] + closed = [{"id": "FPLAN-0002", "subject": "Done"}] + result = mod._build_section_data(active, closed, 10) + + assert result["managed_by"] == "flow" + assert result["active_plans"] == active + assert result["active_count"] == 1 + assert result["recently_closed"] == closed + assert result["total_plans"] == 10 + + def test_empty_lists(self): + """Handles empty active and closed lists.""" + mod = _import_mod() + result = mod._build_section_data([], [], 0) + assert result["active_count"] == 0 + assert result["active_plans"] == [] + assert result["recently_closed"] == [] + assert result["total_plans"] == 0 + + +# ═══════════════════════════════════════════════════════════ +# 8. push_flow_to_branch_dashboard +# ═══════════════════════════════════════════════════════════ + + +class TestPushFlowToBranchDashboard: + """Tests for push_flow_to_branch_dashboard.""" + + def test_returns_false_if_no_dashboard_exists(self, tmp_path): + """Returns False when DASHBOARD.local.json does not exist.""" + mod = _import_mod() + result = mod.push_flow_to_branch_dashboard(tmp_path) + assert result is False + + def test_success_calls_log_operation(self, tmp_path, mock_json_handler): + """Logs via json_handler on successful push.""" + mod = _import_mod() + dashboard_path = tmp_path / "DASHBOARD.local.json" + dashboard_path.write_text(json.dumps({"sections": {}}), encoding="utf-8") + + mock_registry = {"plans": {}, "next_number": 1} + with patch.object(mod, "_load_registry", return_value=mock_registry): + result = mod.push_flow_to_branch_dashboard(tmp_path) + + assert result is True + mock_json_handler.assert_called_once() + call_args = mock_json_handler.call_args + assert call_args[0][0] == "branch_dashboard_pushed" + assert call_args[0][1]["success"] is True + + def test_orchestrates_full_pipeline(self, tmp_path, mock_json_handler): + """Main handler calls load, filter, build, write in sequence.""" + mod = _import_mod() + dashboard_path = tmp_path / "DASHBOARD.local.json" + dashboard_path.write_text(json.dumps({"sections": {}}), encoding="utf-8") + + mock_registry = { + "plans": { + "1": { + "subject": "Active plan", + "status": "open", + "created": "2026-04-20", + "file_path": str(tmp_path / "FPLAN-0001_active.md"), + "location": str(tmp_path), + }, + }, + "next_number": 2, + } + with patch.object(mod, "_load_registry", return_value=mock_registry): + result = mod.push_flow_to_branch_dashboard(tmp_path) + + assert result is True + updated = json.loads(dashboard_path.read_text(encoding="utf-8")) + flow_section = updated["sections"]["flow"] + assert flow_section["active_count"] == 1 + assert flow_section["active_plans"][0]["id"] == "FPLAN-0001" + + def test_returns_false_on_exception(self, tmp_path): + """Returns False when an exception occurs in the main handler.""" + mod = _import_mod() + dashboard_path = tmp_path / "DASHBOARD.local.json" + dashboard_path.write_text(json.dumps({"sections": {}}), encoding="utf-8") + + with patch.object(mod, "_load_registry", side_effect=RuntimeError("registry exploded")): + result = mod.push_flow_to_branch_dashboard(tmp_path) + + assert result is False + + def test_returns_false_when_write_section_fails(self, tmp_path): + """Returns False when _write_dashboard_section fails.""" + mod = _import_mod() + dashboard_path = tmp_path / "DASHBOARD.local.json" + dashboard_path.write_text(json.dumps({"sections": {}}), encoding="utf-8") + + mock_registry = {"plans": {}, "next_number": 1} + with ( + patch.object(mod, "_load_registry", return_value=mock_registry), + patch.object(mod, "_write_dashboard_section", return_value=False), + ): + result = mod.push_flow_to_branch_dashboard(tmp_path) + + assert result is False diff --git a/src/aipass/flow/tests/test_registry_ops.py b/src/aipass/flow/tests/test_registry_ops.py new file mode 100644 index 00000000..920588e9 --- /dev/null +++ b/src/aipass/flow/tests/test_registry_ops.py @@ -0,0 +1,936 @@ +# =================== AIPass ==================== +# Name: test_registry_ops.py +# Description: Tests for registry_ops handler — template registry CRUD +# Version: 1.0.0 +# Created: 2026-04-26 +# Modified: 2026-04-26 +# ============================================= + +"""Tests for registry_ops: template registry CRUD, auto-healing, discovery, edge cases.""" + +import json +from pathlib import Path +from unittest.mock import patch + +import pytest + +# --------------------------------------------------------------------------- +# Module-level patch targets (patch where used, not where defined) +# --------------------------------------------------------------------------- + +_MOD = "aipass.flow.apps.handlers.template.registry_ops" + + +def _import_mod(): + import aipass.flow.apps.handlers.template.registry_ops as mod + + return mod + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def setup_flow_root(tmp_path, monkeypatch): + """Redirect FLOW_ROOT and REGISTRY_PATH to tmp_path for isolation.""" + mod = _import_mod() + flow_root = tmp_path / "flow" + flow_root.mkdir() + (flow_root / "flow_json").mkdir() + (flow_root / "templates").mkdir() + monkeypatch.setattr(mod, "FLOW_ROOT", flow_root) + monkeypatch.setattr(mod, "REGISTRY_PATH", flow_root / "flow_json" / "template_registry.json") + return flow_root + + +def _write_registry(flow_root: Path, data: dict) -> Path: + """Helper: write a template registry JSON and return its path.""" + path = flow_root / "flow_json" / "template_registry.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(data, indent=2), encoding="utf-8") + return path + + +def _valid_registry(extra_types: dict | None = None) -> dict: + """Return a minimal valid registry dict with default types.""" + types = { + "flow_plans": { + "prefix": "FPLAN", + "shorthand": "fplan", + "created": "2026-03-18", + "registered_by": "system", + }, + "dev_plans": { + "prefix": "DPLAN", + "shorthand": "dplan", + "created": "2026-03-18", + "registered_by": "system", + }, + } + if extra_types: + types.update(extra_types) + return { + "types": types, + "metadata": { + "version": "1.0.0", + "last_updated": "2026-03-18", + "type_count": len(types), + }, + } + + +def _create_template_dir(flow_root: Path, name: str, md_files: list[str] | None = None) -> Path: + """Create a template directory under templates/ with optional .md files.""" + tpl_dir = flow_root / "templates" / name + tpl_dir.mkdir(parents=True, exist_ok=True) + for md in md_files or []: + (tpl_dir / md).write_text(f"# {md}", encoding="utf-8") + return tpl_dir + + +# ============================================================================= +# load_registry +# ============================================================================= + + +class TestLoadRegistry: + """Tests for load_registry() — auto-creation, healing, corrupt handling.""" + + def test_creates_registry_when_missing(self, setup_flow_root): + """Auto-creates registry file with defaults when it does not exist.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + + result = mod.load_registry() + + assert "types" in result + assert "metadata" in result + assert "flow_plans" in result["types"] + assert "dev_plans" in result["types"] + reg_path = setup_flow_root / "flow_json" / "template_registry.json" + assert reg_path.exists() + + def test_loads_existing_valid_registry(self, setup_flow_root): + """Loads a valid existing registry from disk.""" + mod = _import_mod() + data = _valid_registry() + _write_registry(setup_flow_root, data) + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + + result = mod.load_registry() + + assert result["types"]["flow_plans"]["prefix"] == "FPLAN" + assert result["types"]["dev_plans"]["prefix"] == "DPLAN" + + def test_corrupt_json_recreates(self, setup_flow_root): + """Corrupt JSON triggers recreation with defaults.""" + mod = _import_mod() + reg_path = setup_flow_root / "flow_json" / "template_registry.json" + reg_path.write_text("{invalid json!!!", encoding="utf-8") + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + + result = mod.load_registry() + + assert "types" in result + assert "flow_plans" in result["types"] + + def test_non_dict_recreates(self, setup_flow_root): + """Non-dict JSON (e.g. a list) triggers recreation.""" + mod = _import_mod() + reg_path = setup_flow_root / "flow_json" / "template_registry.json" + reg_path.write_text(json.dumps([1, 2, 3]), encoding="utf-8") + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + + result = mod.load_registry() + + assert isinstance(result, dict) + assert "types" in result + + def test_heals_missing_types_key(self, setup_flow_root): + """Auto-heals missing 'types' key by injecting defaults.""" + mod = _import_mod() + data = {"metadata": {"version": "1.0.0", "last_updated": "2026-01-01", "type_count": 0}} + _write_registry(setup_flow_root, data) + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + + result = mod.load_registry() + + assert "types" in result + assert "flow_plans" in result["types"] + assert "dev_plans" in result["types"] + + def test_heals_missing_metadata_key(self, setup_flow_root): + """Auto-heals missing 'metadata' key.""" + mod = _import_mod() + data = { + "types": { + "flow_plans": {"prefix": "FPLAN", "shorthand": "fplan"}, + "dev_plans": {"prefix": "DPLAN", "shorthand": "dplan"}, + } + } + _write_registry(setup_flow_root, data) + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + + result = mod.load_registry() + + assert "metadata" in result + assert "version" in result["metadata"] + assert "type_count" in result["metadata"] + + def test_calls_prune_and_auto_register(self, setup_flow_root): + """load_registry invokes prune and auto-register.""" + mod = _import_mod() + data = _valid_registry() + _write_registry(setup_flow_root, data) + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + + with ( + patch(f"{_MOD}._prune_orphaned_types", return_value=False) as mock_prune, + patch(f"{_MOD}._auto_register_new_types", return_value=False) as mock_auto, + ): + mod.load_registry() + mock_prune.assert_called_once() + mock_auto.assert_called_once() + + +# ============================================================================= +# save_registry +# ============================================================================= + + +class TestSaveRegistry: + """Tests for save_registry() — writing, metadata update, validation.""" + + def test_saves_valid_registry(self, setup_flow_root): + """Writes valid registry JSON to disk.""" + mod = _import_mod() + data = _valid_registry() + + result = mod.save_registry(data) + + assert result is True + reg_path = setup_flow_root / "flow_json" / "template_registry.json" + assert reg_path.exists() + saved = json.loads(reg_path.read_text(encoding="utf-8")) + assert "types" in saved + assert "metadata" in saved + + def test_updates_metadata_on_save(self, setup_flow_root): + """Updates last_updated and type_count in metadata.""" + mod = _import_mod() + data = _valid_registry() + data["metadata"]["last_updated"] = "1999-01-01" + data["metadata"]["type_count"] = 0 + + mod.save_registry(data) + + reg_path = setup_flow_root / "flow_json" / "template_registry.json" + saved = json.loads(reg_path.read_text(encoding="utf-8")) + assert saved["metadata"]["last_updated"] != "1999-01-01" + assert saved["metadata"]["type_count"] == 2 + + def test_returns_false_for_invalid_structure_not_dict(self, setup_flow_root): + """Returns False when data is not a dict.""" + mod = _import_mod() + + result = mod.save_registry("not a dict") # type: ignore[arg-type] + + assert result is False + + def test_returns_false_for_missing_types_key(self, setup_flow_root): + """Returns False when 'types' key is absent.""" + mod = _import_mod() + + result = mod.save_registry({"metadata": {}}) + + assert result is False + + def test_returns_false_on_os_error(self, setup_flow_root, monkeypatch): + """Returns False when file write fails with OSError.""" + mod = _import_mod() + data = _valid_registry() + monkeypatch.setattr(mod, "REGISTRY_PATH", Path("/proc/nonexistent/registry.json")) + + result = mod.save_registry(data) + + assert result is False + + def test_logs_via_json_handler(self, setup_flow_root, mock_json_handler): + """Calls json_handler.log_operation on successful save.""" + mod = _import_mod() + data = _valid_registry() + + mod.save_registry(data) + + mock_json_handler.assert_called() + + +# ============================================================================= +# add_type +# ============================================================================= + + +class TestAddType: + """Tests for add_type() — validation, registration, plan registry creation.""" + + def test_adds_new_type_successfully(self, setup_flow_root): + """Registers a new type when all validations pass.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "task_plans", ["task_template.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + # Prevent auto-register from claiming task_plans before add_type does + with patch(f"{_MOD}._auto_register_new_types", return_value=False): + result = mod.add_type("task_plans", "TPLAN", "test") + + assert result is True + reg = mod.load_registry() + assert "task_plans" in reg["types"] + assert reg["types"]["task_plans"]["prefix"] == "TPLAN" + + def test_rejects_duplicate_dir_name(self, setup_flow_root): + """Returns False if dir_name is already registered.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + result = mod.add_type("flow_plans", "XPLAN", "test") + + assert result is False + + def test_rejects_duplicate_prefix_case_insensitive(self, setup_flow_root): + """Returns False if prefix already taken (case-insensitive).""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "task_plans", ["template.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + result = mod.add_type("task_plans", "fplan", "test") + + assert result is False + + def test_rejects_missing_template_dir(self, setup_flow_root): + """Returns False if template directory does not exist.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + result = mod.add_type("nonexistent_plans", "NPLAN", "test") + + assert result is False + + def test_rejects_dir_without_md_files(self, setup_flow_root): + """Returns False if template directory has no .md files.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "empty_plans") + data = _valid_registry() + _write_registry(setup_flow_root, data) + + result = mod.add_type("empty_plans", "EPLAN", "test") + + assert result is False + + def test_creates_plan_registry_on_success(self, setup_flow_root): + """Creates plan registry JSON for the new type.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "task_plans", ["task.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + mod.add_type("task_plans", "TPLAN", "test") + + plan_reg = setup_flow_root / "flow_json" / "tplan_registry.json" + assert plan_reg.exists() + content = json.loads(plan_reg.read_text(encoding="utf-8")) + assert content["next_number"] == 1 + assert content["plans"] == {} + + +# ============================================================================= +# remove_type +# ============================================================================= + + +class TestRemoveType: + """Tests for remove_type() — removal, protection, not-found.""" + + def test_removes_existing_type(self, setup_flow_root): + """Successfully removes a non-protected type.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "task_plans", ["task.md"]) + extra = { + "task_plans": { + "prefix": "TPLAN", + "shorthand": "tplan", + "created": "2026-04-01", + "registered_by": "test", + } + } + data = _valid_registry(extra_types=extra) + _write_registry(setup_flow_root, data) + + # Prevent auto-register from re-adding task_plans after removal + with patch(f"{_MOD}._auto_register_new_types", return_value=False): + result = mod.remove_type("task_plans") + + assert result is True + reg = mod.load_registry() + assert "task_plans" not in reg["types"] + + def test_returns_false_if_not_found(self, setup_flow_root): + """Returns False if dir_name is not in registry.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + result = mod.remove_type("nonexistent_plans") + + assert result is False + + def test_returns_false_for_protected_flow_plans(self, setup_flow_root): + """Cannot remove protected type flow_plans.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + result = mod.remove_type("flow_plans") + + assert result is False + + def test_returns_false_for_protected_dev_plans(self, setup_flow_root): + """Cannot remove protected type dev_plans.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + result = mod.remove_type("dev_plans") + + assert result is False + + +# ============================================================================= +# prefix_exists +# ============================================================================= + + +class TestPrefixExists: + """Tests for prefix_exists() — case-insensitive prefix lookup.""" + + def test_finds_existing_prefix_exact_case(self, setup_flow_root): + """Finds prefix with exact case match.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + assert mod.prefix_exists("FPLAN") is True + + def test_finds_existing_prefix_case_insensitive(self, setup_flow_root): + """Finds prefix regardless of case.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + assert mod.prefix_exists("fplan") is True + assert mod.prefix_exists("Fplan") is True + + def test_returns_false_for_unknown_prefix(self, setup_flow_root): + """Returns False for a prefix not in the registry.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + assert mod.prefix_exists("ZPLAN") is False + + +# ============================================================================= +# get_prefix_map +# ============================================================================= + + +class TestGetPrefixMap: + """Tests for get_prefix_map() — dir_name to prefix mapping.""" + + def test_returns_all_registered_prefixes(self, setup_flow_root): + """Returns {dir_name: prefix} for all types.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + result = mod.get_prefix_map() + + assert result["flow_plans"] == "FPLAN" + assert result["dev_plans"] == "DPLAN" + + def test_includes_custom_types(self, setup_flow_root): + """Includes custom registered types in the map.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "task_plans", ["task.md"]) + extra = { + "task_plans": { + "prefix": "TPLAN", + "shorthand": "tplan", + "created": "2026-04-01", + "registered_by": "test", + } + } + data = _valid_registry(extra_types=extra) + _write_registry(setup_flow_root, data) + + result = mod.get_prefix_map() + + assert result["task_plans"] == "TPLAN" + assert len(result) == 3 + + +# ============================================================================= +# get_type_map +# ============================================================================= + + +class TestGetTypeMap: + """Tests for get_type_map() — shorthand to dir_name mapping.""" + + def test_returns_default_entry(self, setup_flow_root): + """Always includes 'default': 'flow_plans'.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + result = mod.get_type_map() + + assert result["default"] == "flow_plans" + + def test_returns_shorthand_mappings(self, setup_flow_root): + """Maps shorthand to dir_name for all types.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + result = mod.get_type_map() + + assert result["fplan"] == "flow_plans" + assert result["dplan"] == "dev_plans" + + def test_includes_custom_type_shorthand(self, setup_flow_root): + """Custom types are included with their shorthand.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "task_plans", ["task.md"]) + extra = { + "task_plans": { + "prefix": "TPLAN", + "shorthand": "tplan", + "created": "2026-04-01", + "registered_by": "test", + } + } + data = _valid_registry(extra_types=extra) + _write_registry(setup_flow_root, data) + + result = mod.get_type_map() + + assert result["tplan"] == "task_plans" + + +# ============================================================================= +# scan_unregistered +# ============================================================================= + + +class TestScanUnregistered: + """Tests for scan_unregistered() — discovery of unregistered template dirs.""" + + def test_finds_unregistered_dir(self, setup_flow_root): + """Detects a template dir not in the registry.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "task_plans", ["task.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + with patch(f"{_MOD}._auto_register_new_types", return_value=False): + result = mod.scan_unregistered() + + names = [r["dir_name"] for r in result] + assert "task_plans" in names + + def test_skips_hidden_dirs(self, setup_flow_root): + """Directories starting with '.' are ignored.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, ".hidden_plans", ["hidden.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + with patch(f"{_MOD}._auto_register_new_types", return_value=False): + result = mod.scan_unregistered() + + names = [r["dir_name"] for r in result] + assert ".hidden_plans" not in names + + def test_skips_underscore_dirs(self, setup_flow_root): + """Directories starting with '_' are ignored.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "_private_plans", ["private.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + with patch(f"{_MOD}._auto_register_new_types", return_value=False): + result = mod.scan_unregistered() + + names = [r["dir_name"] for r in result] + assert "_private_plans" not in names + + def test_skips_dirs_without_md_files(self, setup_flow_root): + """Dirs with no .md files are not returned.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "empty_plans") + data = _valid_registry() + _write_registry(setup_flow_root, data) + + with patch(f"{_MOD}._auto_register_new_types", return_value=False): + result = mod.scan_unregistered() + + names = [r["dir_name"] for r in result] + assert "empty_plans" not in names + + def test_returns_empty_when_all_registered(self, setup_flow_root): + """Returns empty list when no unregistered dirs exist.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + with patch(f"{_MOD}._auto_register_new_types", return_value=False): + result = mod.scan_unregistered() + + assert result == [] + + def test_returns_template_metadata(self, setup_flow_root): + """Each result includes dir_name, template_count, and templates.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "audit_plans", ["audit_a.md", "audit_b.md"]) + data = _valid_registry() + _write_registry(setup_flow_root, data) + + with patch(f"{_MOD}._auto_register_new_types", return_value=False): + result = mod.scan_unregistered() + + assert len(result) == 1 + entry = result[0] + assert entry["dir_name"] == "audit_plans" + assert entry["template_count"] == 2 + templates = entry["templates"] + assert isinstance(templates, list) + assert sorted(templates) == ["audit_a", "audit_b"] + + def test_returns_empty_when_templates_dir_missing(self, setup_flow_root): + """Returns empty list when templates/ directory does not exist.""" + import shutil + + mod = _import_mod() + data = _valid_registry() + _write_registry(setup_flow_root, data) + tpl_dir = setup_flow_root / "templates" + if tpl_dir.exists(): + shutil.rmtree(tpl_dir) + + with patch(f"{_MOD}._prune_orphaned_types", return_value=False): + with patch(f"{_MOD}._auto_register_new_types", return_value=False): + result = mod.scan_unregistered() + + assert result == [] + + +# ============================================================================= +# _prune_orphaned_types +# ============================================================================= + + +class TestPruneOrphanedTypes: + """Tests for _prune_orphaned_types() — cleaning stale entries.""" + + def test_prunes_orphaned_non_protected_type(self, setup_flow_root): + """Removes entries whose template directory is missing.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + extra = { + "task_plans": { + "prefix": "TPLAN", + "shorthand": "tplan", + "created": "2026-04-01", + "registered_by": "test", + } + } + data = _valid_registry(extra_types=extra) + + result = mod._prune_orphaned_types(data) + + assert result is True + assert "task_plans" not in data["types"] + + def test_does_not_prune_protected_types(self, setup_flow_root): + """Protected types (flow_plans, dev_plans) survive even without dirs.""" + mod = _import_mod() + data = _valid_registry() + + mod._prune_orphaned_types(data) + + assert "flow_plans" in data["types"] + assert "dev_plans" in data["types"] + + def test_returns_false_when_nothing_to_prune(self, setup_flow_root): + """Returns False when all non-protected types have directories.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + data = _valid_registry() + + result = mod._prune_orphaned_types(data) + + assert result is False + + def test_deletes_orphan_plan_registry_json(self, setup_flow_root): + """Deletes the associated plan registry JSON when pruning.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + extra = { + "task_plans": { + "prefix": "TPLAN", + "shorthand": "tplan", + "created": "2026-04-01", + "registered_by": "test", + } + } + data = _valid_registry(extra_types=extra) + plan_reg = setup_flow_root / "flow_json" / "tplan_registry.json" + plan_reg.write_text(json.dumps({"next_number": 1, "plans": {}}), encoding="utf-8") + + mod._prune_orphaned_types(data) + + assert not plan_reg.exists() + + +# ============================================================================= +# _auto_register_new_types +# ============================================================================= + + +class TestAutoRegisterNewTypes: + """Tests for _auto_register_new_types() — automatic discovery and registration.""" + + def test_registers_new_template_dir(self, setup_flow_root): + """Finds and registers a new template directory.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "task_plans", ["task.md"]) + data = _valid_registry() + + result = mod._auto_register_new_types(data) + + assert result is True + assert "task_plans" in data["types"] + assert data["types"]["task_plans"]["registered_by"] == "auto" + + def test_skips_hidden_dirs(self, setup_flow_root): + """Directories starting with '.' are skipped.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, ".hidden", ["secret.md"]) + data = _valid_registry() + + mod._auto_register_new_types(data) + + assert ".hidden" not in data["types"] + + def test_skips_pycache(self, setup_flow_root): + """__pycache__ directories are skipped.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "__pycache__", ["cache.md"]) + data = _valid_registry() + + mod._auto_register_new_types(data) + + assert "__pycache__" not in data["types"] + + def test_skips_underscore_dirs(self, setup_flow_root): + """Directories starting with '_' are skipped.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "_internal", ["internal.md"]) + data = _valid_registry() + + mod._auto_register_new_types(data) + + assert "_internal" not in data["types"] + + def test_returns_false_when_templates_dir_missing(self, setup_flow_root): + """Returns False when templates/ directory does not exist.""" + import shutil + + mod = _import_mod() + data = _valid_registry() + tpl_dir = setup_flow_root / "templates" + if tpl_dir.exists(): + shutil.rmtree(tpl_dir) + + result = mod._auto_register_new_types(data) + + assert result is False + + def test_returns_false_when_nothing_new(self, setup_flow_root): + """Returns False when all dirs are already registered.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + data = _valid_registry() + + result = mod._auto_register_new_types(data) + + assert result is False + + def test_handles_prefix_collision_single_char(self, setup_flow_root): + """Skips registration when derived prefix collides and no fallback.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + # "f" would try FPLAN (collision), but len("f") == 1 so no 2-char fallback + _create_template_dir(setup_flow_root, "f", ["test.md"]) + data = _valid_registry() + + mod._auto_register_new_types(data) + + assert "f" not in data["types"] + + def test_creates_plan_registry_for_auto_registered(self, setup_flow_root): + """Auto-registered types get a plan registry JSON created.""" + mod = _import_mod() + _create_template_dir(setup_flow_root, "flow_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "dev_plans", ["default.md"]) + _create_template_dir(setup_flow_root, "task_plans", ["task.md"]) + data = _valid_registry() + + mod._auto_register_new_types(data) + + plan_reg = setup_flow_root / "flow_json" / "tplan_registry.json" + assert plan_reg.exists() + content = json.loads(plan_reg.read_text(encoding="utf-8")) + assert content["next_number"] == 1 + + +# ============================================================================= +# _derive_prefix +# ============================================================================= + + +class TestDerivePrefix: + """Tests for _derive_prefix() — prefix derivation and collision handling.""" + + def test_basic_derivation(self): + """Derives first letter + 'PLAN' from dir name.""" + mod = _import_mod() + + result = mod._derive_prefix("task_plans", set()) + + assert result == "TPLAN" + + def test_collision_falls_back_to_two_chars(self): + """On collision, tries first two letters + 'PLAN'.""" + mod = _import_mod() + + result = mod._derive_prefix("task_plans", {"TPLAN"}) + + assert result == "TAPLAN" + + def test_returns_none_on_double_collision(self): + """Returns None when both single and double-char prefix collide.""" + mod = _import_mod() + + result = mod._derive_prefix("task_plans", {"TPLAN", "TAPLAN"}) + + assert result is None + + def test_single_char_dir_no_fallback(self): + """Single-char dir name cannot fall back to 2-char prefix.""" + mod = _import_mod() + + result = mod._derive_prefix("t", {"TPLAN"}) + + assert result is None + + def test_empty_dir_name_gives_xplan(self): + """Empty dir name (empty first_word) falls back to XPLAN.""" + mod = _import_mod() + + result = mod._derive_prefix("", set()) + + assert result == "XPLAN" + + def test_underscore_prefix_splits_on_underscore(self): + """Uses only the first word before underscore.""" + mod = _import_mod() + + result = mod._derive_prefix("security_audit_plans", set()) + + assert result == "SPLAN" diff --git a/src/aipass/flow/tests/test_update_local.py b/src/aipass/flow/tests/test_update_local.py new file mode 100644 index 00000000..9c301425 --- /dev/null +++ b/src/aipass/flow/tests/test_update_local.py @@ -0,0 +1,725 @@ +# =================== AIPass ==================== +# Name: test_update_local.py +# Description: Tests for update_local handler — Flow dashboard updates +# Version: 1.0.0 +# Created: 2026-04-26 +# Modified: 2026-04-26 +# ============================================= + +"""Tests for update_local handler — Flow dashboard updates.""" + +import json +from pathlib import Path +import pytest + + +# ─── Patch targets ─────────────────────────────────────── +_MOD = "aipass.flow.apps.handlers.dashboard.update_local" + + +def _import_mod(): + """Import update_local module and return it.""" + import aipass.flow.apps.handlers.dashboard.update_local as mod + + return mod + + +# ─── Shared fixtures ───────────────────────────────────── + + +@pytest.fixture +def setup_paths(tmp_path, monkeypatch): + """Redirect all module-level path constants into tmp_path.""" + mod = _import_mod() + flow_root = tmp_path / "flow" + flow_root.mkdir() + flow_json = flow_root / "flow_json" + flow_json.mkdir() + monkeypatch.setattr(mod, "FLOW_ROOT", flow_root) + monkeypatch.setattr(mod, "FLOW_JSON_DIR", flow_json) + monkeypatch.setattr(mod, "REGISTRY_FILE", flow_json / "fplan_registry.json") + monkeypatch.setattr(mod, "DASHBOARD_FILE", flow_root / "DASHBOARD.local.json") + return flow_root + + +def _write_json(path: Path, data: dict) -> None: + """Helper to write JSON to a file.""" + path.write_text(json.dumps(data, indent=2), encoding="utf-8") + + +def _read_json(path: Path) -> dict: + """Helper to read JSON from a file.""" + return json.loads(path.read_text(encoding="utf-8")) + + +def _make_template_registry(flow_json: Path, types: dict | None = None) -> Path: + """Create a template_registry.json with the given types.""" + if types is None: + types = { + "flow_plans": {"prefix": "FPLAN", "shorthand": "fplan"}, + "dev_plans": {"prefix": "DPLAN", "shorthand": "dplan"}, + "test_plans": {"prefix": "TDPLAN", "shorthand": "tdplan"}, + } + path = flow_json / "template_registry.json" + _write_json(path, {"types": types}) + return path + + +def _make_registry(flow_json: Path, filename: str, plans: dict, next_number: int = 10) -> Path: + """Create a registry file with plans.""" + path = flow_json / filename + _write_json(path, {"plans": plans, "next_number": next_number}) + return path + + +# ═══════════════════════════════════════════════════════════ +# 1. _get_all_registry_files +# ═══════════════════════════════════════════════════════════ + + +class TestGetAllRegistryFiles: + """Tests for _get_all_registry_files — template registry parsing.""" + + def test_returns_filenames_from_template_registry(self, setup_paths): + """Should return per-type registry filenames when template exists.""" + mod = _import_mod() + flow_json = setup_paths / "flow_json" + _make_template_registry(flow_json) + result = mod._get_all_registry_files() + assert "fplan_registry.json" in result + assert "dplan_registry.json" in result + assert "tdplan_registry.json" in result + + def test_no_duplicates_in_result(self, setup_paths): + """Should not produce duplicate filenames.""" + mod = _import_mod() + flow_json = setup_paths / "flow_json" + types = { + "a": {"prefix": "FPLAN"}, + "b": {"prefix": "FPLAN"}, + } + _make_template_registry(flow_json, types) + result = mod._get_all_registry_files() + assert result.count("fplan_registry.json") == 1 + + def test_falls_back_when_template_missing(self, setup_paths): + """Should return default REGISTRY_FILE.name when template is absent.""" + mod = _import_mod() + result = mod._get_all_registry_files() + assert result == [mod.REGISTRY_FILE.name] + + def test_falls_back_on_read_error(self, setup_paths): + """Should return default on JSON decode error.""" + mod = _import_mod() + flow_json = setup_paths / "flow_json" + template_path = flow_json / "template_registry.json" + template_path.write_text("NOT VALID JSON", encoding="utf-8") + result = mod._get_all_registry_files() + assert result == [mod.REGISTRY_FILE.name] + + def test_falls_back_when_no_prefixes_found(self, setup_paths): + """Should return default when types exist but none have a prefix.""" + mod = _import_mod() + flow_json = setup_paths / "flow_json" + types = { + "empty_type": {"shorthand": "nope"}, + "also_empty": {}, + } + _make_template_registry(flow_json, types) + result = mod._get_all_registry_files() + assert result == [mod.REGISTRY_FILE.name] + + def test_skips_entries_with_empty_prefix(self, setup_paths): + """Should skip types whose prefix is an empty string.""" + mod = _import_mod() + flow_json = setup_paths / "flow_json" + types = { + "good": {"prefix": "FPLAN"}, + "bad": {"prefix": ""}, + } + _make_template_registry(flow_json, types) + result = mod._get_all_registry_files() + assert result == ["fplan_registry.json"] + + +# ═══════════════════════════════════════════════════════════ +# 2. _read_registry +# ═══════════════════════════════════════════════════════════ + + +class TestReadRegistry: + """Tests for _read_registry — multi-registry merging.""" + + def test_merges_multiple_registries(self, setup_paths): + """Should merge plans from multiple registry files.""" + mod = _import_mod() + flow_json = setup_paths / "flow_json" + _make_template_registry( + flow_json, + { + "flow": {"prefix": "FPLAN"}, + "dev": {"prefix": "DPLAN"}, + }, + ) + _make_registry( + flow_json, + "fplan_registry.json", + { + "1": {"subject": "Plan A", "status": "open", "location": "flow"}, + }, + next_number=5, + ) + _make_registry( + flow_json, + "dplan_registry.json", + { + "2": {"subject": "Plan B", "status": "closed", "location": "flow"}, + }, + next_number=8, + ) + + result = mod._read_registry() + assert result is not None + assert "1" in result["plans"] + assert "2" in result["plans"] + + def test_keeps_highest_next_number(self, setup_paths): + """Should keep the highest next_number across registries.""" + mod = _import_mod() + flow_json = setup_paths / "flow_json" + _make_template_registry( + flow_json, + { + "a": {"prefix": "FPLAN"}, + "b": {"prefix": "DPLAN"}, + }, + ) + _make_registry(flow_json, "fplan_registry.json", {}, next_number=3) + _make_registry(flow_json, "dplan_registry.json", {}, next_number=15) + + result = mod._read_registry() + assert result is not None + assert result["next_number"] == 15 + + def test_returns_none_if_no_registries_found(self, setup_paths): + """Should return None when no registry files exist on disk.""" + mod = _import_mod() + result = mod._read_registry() + assert result is None + + def test_handles_read_error_gracefully(self, setup_paths): + """Should skip corrupt registry files without crashing.""" + mod = _import_mod() + flow_json = setup_paths / "flow_json" + # No template registry, so it falls back to REGISTRY_FILE.name + corrupt = flow_json / mod.REGISTRY_FILE.name + corrupt.write_text("{bad json", encoding="utf-8") + result = mod._read_registry() + # The file exists but can't be parsed; found_any stays False + assert result is None + + def test_single_valid_registry(self, setup_paths): + """Should work with a single valid registry file.""" + mod = _import_mod() + flow_json = setup_paths / "flow_json" + _make_registry( + flow_json, + mod.REGISTRY_FILE.name, + { + "1": {"subject": "Solo plan", "status": "open", "location": "flow"}, + }, + next_number=2, + ) + + result = mod._read_registry() + assert result is not None + assert result["plans"]["1"]["subject"] == "Solo plan" + assert result["next_number"] == 2 + + +# ═══════════════════════════════════════════════════════════ +# 3. _extract_flow_plans +# ═══════════════════════════════════════════════════════════ + + +class TestExtractFlowPlans: + """Tests for _extract_flow_plans — filtering and partitioning.""" + + def test_filters_plans_by_flow_location(self): + """Should only include plans where location contains 'flow'.""" + mod = _import_mod() + registry = { + "plans": { + "1": {"subject": "A", "status": "open", "location": "flow", "file_path": "FPLAN-0001.md"}, + "2": {"subject": "B", "status": "open", "location": "drone", "file_path": "FPLAN-0002.md"}, + "3": {"subject": "C", "status": "open", "location": "flow/sub", "file_path": "FPLAN-0003.md"}, + } + } + active, closed = mod._extract_flow_plans(registry) + assert len(active) == 2 + plan_ids = [p["plan_id"] for p in active] + assert "FPLAN-1" in plan_ids + assert "FPLAN-3" in plan_ids + + def test_partitions_by_status(self): + """Should separate open and closed plans.""" + mod = _import_mod() + registry = { + "plans": { + "1": {"subject": "Open", "status": "open", "location": "flow", "file_path": "FPLAN-0001.md"}, + "2": {"subject": "Closed", "status": "closed", "location": "flow", "file_path": "FPLAN-0002.md"}, + } + } + active, closed = mod._extract_flow_plans(registry) + assert len(active) == 1 + assert len(closed) == 1 + assert active[0]["status"] == "open" + assert closed[0]["status"] == "closed" + + def test_extracts_prefix_from_file_path(self): + """Should derive the plan prefix from the filename.""" + mod = _import_mod() + registry = { + "plans": { + "4": { + "subject": "Dev", + "status": "open", + "location": "flow", + "file_path": "/some/path/DPLAN-0004_dev_thing.md", + }, + } + } + active, _ = mod._extract_flow_plans(registry) + assert active[0]["plan_id"] == "DPLAN-4" + + def test_extracts_tdplan_prefix(self): + """Should handle TDPLAN prefix correctly.""" + mod = _import_mod() + registry = { + "plans": { + "7": { + "subject": "Test plan", + "status": "open", + "location": "flow", + "file_path": "TDPLAN-0007_test.md", + }, + } + } + active, _ = mod._extract_flow_plans(registry) + assert active[0]["plan_id"] == "TDPLAN-7" + + def test_defaults_to_fplan_when_no_prefix_match(self): + """Should default to FPLAN when filename has no recognizable prefix.""" + mod = _import_mod() + registry = { + "plans": { + "9": { + "subject": "Mystery", + "status": "open", + "location": "flow", + "file_path": "random_file.md", + }, + } + } + active, _ = mod._extract_flow_plans(registry) + assert active[0]["plan_id"] == "FPLAN-9" + + def test_sorts_active_and_closed_by_plan_id(self): + """Should sort both lists by plan_id.""" + mod = _import_mod() + registry = { + "plans": { + "3": {"subject": "C", "status": "open", "location": "flow", "file_path": "FPLAN-0003.md"}, + "1": {"subject": "A", "status": "open", "location": "flow", "file_path": "FPLAN-0001.md"}, + "5": {"subject": "E", "status": "closed", "location": "flow", "file_path": "FPLAN-0005.md"}, + "2": {"subject": "B", "status": "closed", "location": "flow", "file_path": "FPLAN-0002.md"}, + } + } + active, closed = mod._extract_flow_plans(registry) + assert [p["plan_id"] for p in active] == ["FPLAN-1", "FPLAN-3"] + assert [p["plan_id"] for p in closed] == ["FPLAN-2", "FPLAN-5"] + + def test_handles_timestamps(self): + """Should include created and closed timestamps when present.""" + mod = _import_mod() + registry = { + "plans": { + "1": { + "subject": "With timestamps", + "status": "closed", + "location": "flow", + "file_path": "FPLAN-0001.md", + "created": "2026-01-01T00:00:00Z", + "closed": "2026-01-02T00:00:00Z", + "closed_reason": "completed", + }, + } + } + _, closed = mod._extract_flow_plans(registry) + assert closed[0]["created"] == "2026-01-01T00:00:00Z" + assert closed[0]["closed"] == "2026-01-02T00:00:00Z" + assert closed[0]["closed_reason"] == "completed" + + def test_plan_without_timestamps(self): + """Should not include timestamp keys when absent from source data.""" + mod = _import_mod() + registry = { + "plans": { + "1": { + "subject": "No times", + "status": "open", + "location": "flow", + "file_path": "FPLAN-0001.md", + }, + } + } + active, _ = mod._extract_flow_plans(registry) + assert "created" not in active[0] + assert "closed" not in active[0] + + def test_empty_plans_dict(self): + """Should return empty lists when no plans exist.""" + mod = _import_mod() + active, closed = mod._extract_flow_plans({"plans": {}}) + assert active == [] + assert closed == [] + + def test_case_insensitive_location_match(self): + """Should match location 'Flow', 'FLOW', etc.""" + mod = _import_mod() + registry = { + "plans": { + "1": {"subject": "A", "status": "open", "location": "Flow", "file_path": "FPLAN-0001.md"}, + "2": {"subject": "B", "status": "open", "location": "FLOW", "file_path": "FPLAN-0002.md"}, + } + } + active, _ = mod._extract_flow_plans(registry) + assert len(active) == 2 + + +# ═══════════════════════════════════════════════════════════ +# 4. _calculate_statistics +# ═══════════════════════════════════════════════════════════ + + +class TestCalculateStatistics: + """Tests for _calculate_statistics — stats computation.""" + + def test_returns_correct_counts(self): + """Should return active_count, total_closed, and next_number.""" + mod = _import_mod() + active = [{"plan_id": "FPLAN-1"}, {"plan_id": "FPLAN-2"}] + closed = [{"plan_id": "FPLAN-3"}] + registry = {"next_number": 10} + stats = mod._calculate_statistics(active, closed, registry) + assert stats == {"active_count": 2, "total_closed": 1, "next_number": 10} + + def test_empty_lists(self): + """Should handle empty active and closed lists.""" + mod = _import_mod() + stats = mod._calculate_statistics([], [], {"next_number": 1}) + assert stats == {"active_count": 0, "total_closed": 0, "next_number": 1} + + def test_defaults_next_number_to_one(self): + """Should default next_number to 1 when missing from registry.""" + mod = _import_mod() + stats = mod._calculate_statistics([], [], {}) + assert stats["next_number"] == 1 + + +# ═══════════════════════════════════════════════════════════ +# 5. _read_existing_dashboard +# ═══════════════════════════════════════════════════════════ + + +class TestReadExistingDashboard: + """Tests for _read_existing_dashboard — safe file reading.""" + + def test_returns_empty_dict_if_file_missing(self, setup_paths): + """Should return {} when DASHBOARD_FILE does not exist.""" + mod = _import_mod() + result = mod._read_existing_dashboard() + assert result == {} + + def test_returns_empty_dict_for_old_markdown_format(self, setup_paths): + """Should return {} if content starts with warning emoji.""" + mod = _import_mod() + dashboard_file = setup_paths / "DASHBOARD.local.json" + dashboard_file.write_text("⚠️ Old markdown content here", encoding="utf-8") + result = mod._read_existing_dashboard() + assert result == {} + + def test_returns_empty_dict_if_empty_content(self, setup_paths): + """Should return {} if file is empty or whitespace only.""" + mod = _import_mod() + dashboard_file = setup_paths / "DASHBOARD.local.json" + dashboard_file.write_text(" \n ", encoding="utf-8") + result = mod._read_existing_dashboard() + assert result == {} + + def test_handles_corrupt_json(self, setup_paths): + """Should return {} on JSON parse error.""" + mod = _import_mod() + dashboard_file = setup_paths / "DASHBOARD.local.json" + dashboard_file.write_text("{broken json", encoding="utf-8") + result = mod._read_existing_dashboard() + assert result == {} + + def test_returns_parsed_json(self, setup_paths): + """Should return parsed dict for valid JSON.""" + mod = _import_mod() + dashboard_file = setup_paths / "DASHBOARD.local.json" + data = {"branch": "FLOW", "custom_section": {"key": "value"}} + _write_json(dashboard_file, data) + result = mod._read_existing_dashboard() + assert result == data + + +# ═══════════════════════════════════════════════════════════ +# 6. _build_dashboard_data +# ═══════════════════════════════════════════════════════════ + + +class TestBuildDashboardData: + """Tests for _build_dashboard_data — dashboard assembly.""" + + def test_preserves_existing_sections(self): + """Should keep existing data from other branches.""" + mod = _import_mod() + existing = {"other_branch_section": {"plans": [1, 2, 3]}} + result = mod._build_dashboard_data([], [], {"active_count": 0, "total_closed": 0, "next_number": 1}, existing) + assert "other_branch_section" in result + assert result["other_branch_section"] == {"plans": [1, 2, 3]} + + def test_sets_branch_and_last_updated(self): + """Should set branch to FLOW and include last_updated.""" + mod = _import_mod() + result = mod._build_dashboard_data([], [], {"active_count": 0, "total_closed": 0, "next_number": 1}, {}) + assert result["branch"] == "FLOW" + assert "last_updated" in result + + def test_updates_flow_plans_section(self): + """Should populate flow_plans with active, recently_closed, statistics.""" + mod = _import_mod() + active = [{"plan_id": "FPLAN-1"}] + closed = [{"plan_id": "FPLAN-2"}] + stats = {"active_count": 1, "total_closed": 1, "next_number": 3} + result = mod._build_dashboard_data(active, closed, stats, {}) + assert result["flow_plans"]["active"] == active + assert result["flow_plans"]["recently_closed"] == closed + assert result["flow_plans"]["statistics"] == stats + + def test_limits_recently_closed_to_last_five(self): + """Should keep only the last 5 closed plans.""" + mod = _import_mod() + closed = [{"plan_id": f"FPLAN-{i}"} for i in range(10)] + stats = {"active_count": 0, "total_closed": 10, "next_number": 11} + result = mod._build_dashboard_data([], closed, stats, {}) + assert len(result["flow_plans"]["recently_closed"]) == 5 + # Should be the last 5 items + assert result["flow_plans"]["recently_closed"] == closed[5:] + + def test_recently_closed_empty_when_no_closed(self): + """Should return empty list for recently_closed when none exist.""" + mod = _import_mod() + result = mod._build_dashboard_data([], [], {"active_count": 0, "total_closed": 0, "next_number": 1}, {}) + assert result["flow_plans"]["recently_closed"] == [] + + def test_does_not_mutate_existing_dict(self): + """Should not modify the original existing dict.""" + mod = _import_mod() + existing = {"keep": "me"} + original_copy = existing.copy() + mod._build_dashboard_data([], [], {"active_count": 0, "total_closed": 0, "next_number": 1}, existing) + assert existing == original_copy + + +# ═══════════════════════════════════════════════════════════ +# 7. _write_dashboard +# ═══════════════════════════════════════════════════════════ + + +class TestWriteDashboard: + """Tests for _write_dashboard — file writing.""" + + def test_writes_json_to_file(self, setup_paths): + """Should write valid JSON to DASHBOARD_FILE.""" + mod = _import_mod() + data = {"branch": "FLOW", "test": True} + result = mod._write_dashboard(data) + assert result is True + written = _read_json(mod.DASHBOARD_FILE) + assert written == data + + def test_creates_parent_dirs(self, tmp_path, monkeypatch): + """Should create parent directories if they do not exist.""" + mod = _import_mod() + deep_path = tmp_path / "a" / "b" / "c" / "DASHBOARD.local.json" + monkeypatch.setattr(mod, "DASHBOARD_FILE", deep_path) + result = mod._write_dashboard({"branch": "FLOW"}) + assert result is True + assert deep_path.exists() + + def test_returns_false_on_write_error(self, setup_paths, monkeypatch): + """Should return False when writing fails.""" + mod = _import_mod() + monkeypatch.setattr(mod, "DASHBOARD_FILE", Path("/nonexistent/readonly/DASHBOARD.local.json")) + result = mod._write_dashboard({"branch": "FLOW"}) + assert result is False + + +# ═══════════════════════════════════════════════════════════ +# 8. update_dashboard_local (full pipeline) +# ═══════════════════════════════════════════════════════════ + + +class TestUpdateDashboardLocal: + """Tests for update_dashboard_local — end-to-end pipeline.""" + + def test_full_pipeline_success(self, setup_paths): + """Should read registry, extract, compute stats, and write dashboard.""" + mod = _import_mod() + flow_json = setup_paths / "flow_json" + _make_registry( + flow_json, + mod.REGISTRY_FILE.name, + { + "1": { + "subject": "Active plan", + "status": "open", + "location": "flow", + "file_path": "FPLAN-0001_active.md", + "created": "2026-04-01", + }, + "2": { + "subject": "Closed plan", + "status": "closed", + "location": "flow", + "file_path": "FPLAN-0002_closed.md", + "created": "2026-03-15", + "closed": "2026-03-20", + "closed_reason": "done", + }, + }, + next_number=5, + ) + + result = mod.update_dashboard_local() + assert result is True + + dashboard = _read_json(mod.DASHBOARD_FILE) + assert dashboard["branch"] == "FLOW" + assert len(dashboard["flow_plans"]["active"]) == 1 + assert len(dashboard["flow_plans"]["recently_closed"]) == 1 + assert dashboard["flow_plans"]["statistics"]["active_count"] == 1 + assert dashboard["flow_plans"]["statistics"]["total_closed"] == 1 + assert dashboard["flow_plans"]["statistics"]["next_number"] == 5 + + def test_returns_false_when_registry_is_none(self, setup_paths): + """Should return False when no registry files exist.""" + mod = _import_mod() + result = mod.update_dashboard_local() + assert result is False + + def test_logs_via_json_handler_on_success(self, setup_paths, mock_json_handler): + """Should call json_handler.log_operation on success.""" + mod = _import_mod() + flow_json = setup_paths / "flow_json" + _make_registry( + flow_json, + mod.REGISTRY_FILE.name, + { + "1": {"subject": "P", "status": "open", "location": "flow", "file_path": "FPLAN-0001.md"}, + }, + next_number=2, + ) + + mod.update_dashboard_local() + mock_json_handler.assert_called_once() + call_args = mock_json_handler.call_args + assert call_args[0][0] == "dashboard_local_updated" + assert call_args[0][1]["success"] is True + + def test_preserves_other_sections_in_existing_dashboard(self, setup_paths): + """Should preserve non-flow sections from existing dashboard.""" + mod = _import_mod() + flow_json = setup_paths / "flow_json" + _make_registry( + flow_json, + mod.REGISTRY_FILE.name, + { + "1": {"subject": "P", "status": "open", "location": "flow", "file_path": "FPLAN-0001.md"}, + }, + next_number=2, + ) + + # Pre-populate dashboard with another branch's section + existing = {"drone_plans": {"active": [{"plan_id": "DPLAN-99"}]}} + _write_json(mod.DASHBOARD_FILE, existing) + + mod.update_dashboard_local() + dashboard = _read_json(mod.DASHBOARD_FILE) + assert "drone_plans" in dashboard + assert dashboard["drone_plans"]["active"][0]["plan_id"] == "DPLAN-99" + + def test_does_not_log_on_failure(self, setup_paths, mock_json_handler): + """Should not call json_handler when pipeline fails.""" + mod = _import_mod() + # No registry files exist => returns False + mod.update_dashboard_local() + mock_json_handler.assert_not_called() + + def test_pipeline_with_multiple_registries(self, setup_paths): + """Should merge plans from multiple registry types.""" + mod = _import_mod() + flow_json = setup_paths / "flow_json" + _make_template_registry( + flow_json, + { + "flow": {"prefix": "FPLAN"}, + "dev": {"prefix": "DPLAN"}, + }, + ) + _make_registry( + flow_json, + "fplan_registry.json", + { + "1": {"subject": "Flow plan", "status": "open", "location": "flow", "file_path": "FPLAN-0001.md"}, + }, + next_number=3, + ) + _make_registry( + flow_json, + "dplan_registry.json", + { + "2": {"subject": "Dev plan", "status": "open", "location": "flow", "file_path": "DPLAN-0002.md"}, + }, + next_number=7, + ) + + result = mod.update_dashboard_local() + assert result is True + + dashboard = _read_json(mod.DASHBOARD_FILE) + assert len(dashboard["flow_plans"]["active"]) == 2 + assert dashboard["flow_plans"]["statistics"]["next_number"] == 7 + + def test_pipeline_filters_non_flow_plans(self, setup_paths): + """Should exclude plans located in other branches.""" + mod = _import_mod() + flow_json = setup_paths / "flow_json" + _make_registry( + flow_json, + mod.REGISTRY_FILE.name, + { + "1": {"subject": "Flow plan", "status": "open", "location": "flow", "file_path": "FPLAN-0001.md"}, + "2": {"subject": "Drone plan", "status": "open", "location": "drone", "file_path": "FPLAN-0002.md"}, + "3": {"subject": "Memory plan", "status": "open", "location": "memory", "file_path": "FPLAN-0003.md"}, + }, + next_number=4, + ) + + mod.update_dashboard_local() + dashboard = _read_json(mod.DASHBOARD_FILE) + assert len(dashboard["flow_plans"]["active"]) == 1 + assert dashboard["flow_plans"]["active"][0]["plan_id"] == "FPLAN-1"