From c6d2bcd35313a31a1827c8bab52b7be748a9a6fa Mon Sep 17 00:00:00 2001 From: AIPass Date: Sat, 14 Mar 2026 23:07:06 -0700 Subject: [PATCH] feat(drone): 188-test suite, interactive backup commands, type fix (#46) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(backup): stderr routing, error masking fix, mtime pre-check, ignore patterns split - Route all error/warning prints to stderr via error()/warning() (backup_core, google_drive_sync) - Fix route_command error masking — show real errors instead of "Unknown command" - Add versioned backup mtime pre-check to skip unchanged files - Extract ignore patterns from config_handler to editable JSON (FPLAN-0037) - New: ignore_patterns.py loads from json_templates/ignore_patterns.json - config_handler.py v3.0.0 re-exports for backwards compatibility - 132 patterns, 14/14 tests pass Co-Authored-By: @backup Co-Authored-By: Claude Opus 4.6 (1M context) * feat(prax): dashboard cleanup — remove devpulse section, fix broken paths Phase 1 dashboard cleanup (FPLAN-0040): - Remove devpulse section from template, all hardcoded defaults, and refresh pipeline - Fix broken paths: old aipass_os/dev_central/devpulse → prax via Path(__file__) - Fix registry: BRANCH_REGISTRY.json → AIPASS_REGISTRY.json with repo root walk-up - Fix relative path resolution: registry stores relative paths, now resolved against repo root - Fix agent_status_writer subprocess import: aipass.devpulse → aipass.prax - Update help text and docstrings: drone @devpulse → drone @prax - Add devpulse to DEPRECATED_SECTIONS in template_pusher and differ - Push-template verified: all 15 branches updated, devpulse section removed Co-Authored-By: @prax * feat(drone): add 188-test suite, interactive commands for backup, type fix - Add 5 test files (188 tests) covering executor, resolver, router, registry_handler, and discovery — built via 3-round audit process - Add snapshot/versioned to interactive command tuple for backup Rich output - Fix type error in router_handler.py:119 (command: Optional[str] → str) - Fix pytest.ini inline comments breaking pytest 9.x - Document interactive commands in README Co-Authored-By: @drone Co-Authored-By: Claude Opus 4.6 (1M context) --------- Co-authored-by: @backup Co-authored-by: Claude Opus 4.6 (1M context) Co-authored-by: @prax Co-authored-by: @drone --- src/aipass/drone/README.md | 20 +- src/aipass/drone/apps/drone.py | 2 +- .../drone/apps/handlers/router_handler.py | 2 +- src/aipass/drone/pytest.ini | 8 +- src/aipass/drone/tests/test_discovery.py | 640 ++++++++++++++++++ src/aipass/drone/tests/test_executor.py | 403 +++++++++++ .../drone/tests/test_registry_handler.py | 534 +++++++++++++++ src/aipass/drone/tests/test_resolver.py | 392 +++++++++++ src/aipass/drone/tests/test_router.py | 630 +++++++++++++++++ .../handlers/dashboard/agent_status_writer.py | 4 +- .../apps/handlers/dashboard/operations.py | 15 +- .../prax/apps/handlers/dashboard/refresh.py | 44 +- .../prax/apps/handlers/dashboard/status.py | 28 +- .../handlers/dashboard/template_differ.py | 36 +- .../handlers/dashboard/template_pusher.py | 38 +- src/aipass/prax/apps/modules/dashboard.py | 29 +- .../prax/templates/.dashboard_version.json | 31 +- .../prax/templates/DASHBOARD.template.json | 5 - 18 files changed, 2737 insertions(+), 124 deletions(-) create mode 100644 src/aipass/drone/tests/test_discovery.py create mode 100644 src/aipass/drone/tests/test_executor.py create mode 100644 src/aipass/drone/tests/test_registry_handler.py create mode 100644 src/aipass/drone/tests/test_resolver.py create mode 100644 src/aipass/drone/tests/test_router.py diff --git a/src/aipass/drone/README.md b/src/aipass/drone/README.md index fc5fdd53..18f0b2e2 100644 --- a/src/aipass/drone/README.md +++ b/src/aipass/drone/README.md @@ -99,6 +99,24 @@ drone/ --- +## Interactive Commands + +By default, drone captures subprocess output (`capture_output=True`) with a 30s timeout. This is safe for AI-to-AI routing but strips Rich colors, buffers progress bars, and kills long-running commands. + +Commands in the interactive tuple bypass capture and inherit the terminal directly — enabling live Rich output, colors, and no timeout. Only add commands here when Patrick needs full terminal experience. + +**Current interactive commands** (in `apps/drone.py`): + +| Command | Reason | +|--------------|---------------------------------------------| +| `monitor` | Prax real-time monitoring (live TUI) | +| `snapshot` | Backup snapshot (Rich progress bars) | +| `versioned` | Backup versioned (Rich progress, long-running) | + +To add a new command: edit the `interactive` tuple in `_handle_target()` in `apps/drone.py`. + +--- + ## Integration Points ### Depends On @@ -113,4 +131,4 @@ drone/ --- -**Last Updated:** 2026-03-08 +**Last Updated:** 2026-03-14 diff --git a/src/aipass/drone/apps/drone.py b/src/aipass/drone/apps/drone.py index 58fc6f1b..8bfc031f 100644 --- a/src/aipass/drone/apps/drone.py +++ b/src/aipass/drone/apps/drone.py @@ -210,7 +210,7 @@ def _handle_target(args: List[str]) -> int: cmd_args = rest[1:] # Long-running interactive commands bypass capture + timeout - interactive = command in ("monitor",) + interactive = command in ("monitor", "snapshot", "versioned") try: result = route_command( diff --git a/src/aipass/drone/apps/handlers/router_handler.py b/src/aipass/drone/apps/handlers/router_handler.py index 8d7570c4..d9a9c443 100644 --- a/src/aipass/drone/apps/handlers/router_handler.py +++ b/src/aipass/drone/apps/handlers/router_handler.py @@ -116,5 +116,5 @@ def execute_branch_command( stderr=result.stderr, exit_code=result.exit_code, branch=branch_name, - command=command, + command=command or "", ) diff --git a/src/aipass/drone/pytest.ini b/src/aipass/drone/pytest.ini index c3f21db6..027f04c6 100644 --- a/src/aipass/drone/pytest.ini +++ b/src/aipass/drone/pytest.ini @@ -9,10 +9,10 @@ python_classes = Test* # Command-line options (always applied) addopts = - -v # Verbose output - --tb=short # Short traceback format - --strict-markers # Raise error on unknown markers - -ra # Show summary of all test outcomes + -v + --tb=short + --strict-markers + -ra # Test markers (for categorizing tests) markers = diff --git a/src/aipass/drone/tests/test_discovery.py b/src/aipass/drone/tests/test_discovery.py new file mode 100644 index 00000000..aa6151b9 --- /dev/null +++ b/src/aipass/drone/tests/test_discovery.py @@ -0,0 +1,640 @@ +# =================== AIPass ==================== +# Name: test_discovery.py +# Description: Tests for module and command discovery +# Version: 1.0.0 +# Created: 2026-03-14 +# Modified: 2026-03-14 +# ============================================= + +"""Tests for module and command discovery. + +Covers handler-layer functions (scan_modules_directory, parse_help_for_commands, +get_help, get_module_introspective) and orchestration-layer functions +(discover_modules, get_help, get_system_help) from the discovery module. +""" + +import subprocess +import types +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from aipass.drone.apps.handlers.discovery_handler import ( + HelpResult, + scan_modules_directory, + parse_help_for_commands, + get_help as handler_get_help, + get_entry_point, + discover_modules as handler_discover_modules, +) +from aipass.drone.apps.handlers.exceptions import ( + BranchNotFoundError, + CommandExecutionError, +) +from aipass.drone.apps.handlers.module_registry_handler import ( + get_module_introspective, +) + + +# ============================================================================= +# scan_modules_directory tests +# ============================================================================= + +class TestScanModulesDirectory: + """Tests for scan_modules_directory().""" + + def test_finds_py_files(self, temp_test_dir: Path): + """Should return stems of .py files in apps/modules/.""" + modules_dir = temp_test_dir / "apps" / "modules" + modules_dir.mkdir(parents=True) + (modules_dir / "alpha.py").write_text("# alpha", encoding="utf-8") + (modules_dir / "beta.py").write_text("# beta", encoding="utf-8") + (modules_dir / "gamma.py").write_text("# gamma", encoding="utf-8") + + result = scan_modules_directory(str(temp_test_dir)) + + assert result == ["alpha", "beta", "gamma"] + + def test_skips_init_and_main(self, temp_test_dir: Path): + """Should exclude __init__.py and __main__.py from results.""" + modules_dir = temp_test_dir / "apps" / "modules" + modules_dir.mkdir(parents=True) + (modules_dir / "__init__.py").write_text("", encoding="utf-8") + (modules_dir / "__main__.py").write_text("", encoding="utf-8") + (modules_dir / "real_module.py").write_text("# code", encoding="utf-8") + + result = scan_modules_directory(str(temp_test_dir)) + + assert "__init__" not in result + assert "__main__" not in result + assert result == ["real_module"] + + def test_ignores_pycache_and_non_py(self, temp_test_dir: Path): + """Should ignore __pycache__ directories and non-.py files.""" + modules_dir = temp_test_dir / "apps" / "modules" + modules_dir.mkdir(parents=True) + (modules_dir / "__pycache__").mkdir() + (modules_dir / "__pycache__" / "cached.pyc").write_bytes(b"\x00") + (modules_dir / "notes.txt").write_text("not python", encoding="utf-8") + (modules_dir / "data.json").write_text("{}", encoding="utf-8") + (modules_dir / "valid.py").write_text("# ok", encoding="utf-8") + + result = scan_modules_directory(str(temp_test_dir)) + + assert result == ["valid"] + + def test_returns_empty_for_empty_directory(self, temp_test_dir: Path): + """Should return empty list when apps/modules/ exists but has no .py files.""" + modules_dir = temp_test_dir / "apps" / "modules" + modules_dir.mkdir(parents=True) + + result = scan_modules_directory(str(temp_test_dir)) + + assert result == [] + + def test_returns_empty_when_no_modules_dir(self, temp_test_dir: Path): + """Should return empty list when apps/modules/ does not exist.""" + result = scan_modules_directory(str(temp_test_dir)) + + assert result == [] + + def test_results_are_sorted(self, temp_test_dir: Path): + """Returned module names should be sorted alphabetically.""" + modules_dir = temp_test_dir / "apps" / "modules" + modules_dir.mkdir(parents=True) + for name in ["zebra", "apple", "mango"]: + (modules_dir / f"{name}.py").write_text("", encoding="utf-8") + + result = scan_modules_directory(str(temp_test_dir)) + + assert result == sorted(result) + assert result == ["apple", "mango", "zebra"] + + +# ============================================================================= +# parse_help_for_commands tests +# ============================================================================= + +class TestParseHelpForCommands: + """Tests for parse_help_for_commands().""" + + def test_extracts_commands_from_commands_section(self): + """Should extract command names from a 'Commands:' section.""" + help_text = ( + "Usage: tool [OPTIONS] COMMAND\n" + "\n" + "Commands:\n" + " init Initialize a project\n" + " build Build the project\n" + " deploy Deploy to production\n" + "\n" + "Options:\n" + " --help Show this message\n" + ) + + result = parse_help_for_commands(help_text) + + assert "init" in result + assert "build" in result + assert "deploy" in result + + def test_extracts_from_subcommands_section(self): + """Should recognize 'Subcommands:' as a section marker.""" + help_text = ( + "MyTool v1.0\n" + "\n" + "Subcommands:\n" + " status Show status\n" + " config Configure settings\n" + "\n" + ) + + result = parse_help_for_commands(help_text) + + assert "status" in result + assert "config" in result + + def test_extracts_from_available_commands_section(self): + """Should recognize 'Available Commands:' as a section marker.""" + help_text = ( + "Usage: app\n" + "\n" + "Available Commands:\n" + " run Run the app\n" + " test Run tests\n" + "\n" + ) + + result = parse_help_for_commands(help_text) + + assert "run" in result + assert "test" in result + + def test_skips_options_flags(self): + """Should not include lines starting with - as commands.""" + help_text = ( + "Commands:\n" + " start Start the server\n" + " -v Verbose mode\n" + " --debug Debug mode\n" + ) + + result = parse_help_for_commands(help_text) + + assert "start" in result + assert "-v" not in result + assert "--debug" not in result + + def test_empty_string_returns_empty_list(self): + """Empty help text should return empty list.""" + result = parse_help_for_commands("") + + assert result == [] + + def test_no_commands_section_returns_empty(self): + """Help text without a commands section should return empty list.""" + help_text = ( + "Usage: tool [OPTIONS]\n" + "\n" + "Options:\n" + " --help Show help\n" + " --version Show version\n" + ) + + result = parse_help_for_commands(help_text) + + assert result == [] + + def test_section_ends_at_blank_line(self): + """Commands section should end at the first blank line after it.""" + help_text = ( + "Commands:\n" + " alpha First command\n" + " beta Second command\n" + "\n" + "Not a command section anymore:\n" + " gamma Should not appear\n" + ) + + result = parse_help_for_commands(help_text) + + assert "alpha" in result + assert "beta" in result + # "gamma" is in a new section after the blank line ended the Commands + # section, and "Not a command section anymore:" is not a recognized marker + assert "gamma" not in result + + +# ============================================================================= +# get_entry_point tests +# ============================================================================= + +class TestGetEntryPoint: + """Tests for get_entry_point().""" + + def test_returns_path_when_entry_exists(self, temp_test_dir: Path): + """Should return the entry point Path when apps/{name}.py exists.""" + apps_dir = temp_test_dir / "apps" + apps_dir.mkdir(parents=True) + entry = apps_dir / "mybranch.py" + entry.write_text("# entry", encoding="utf-8") + + result = get_entry_point(str(temp_test_dir), "mybranch") + + assert result is not None + assert result == entry + + def test_returns_none_when_missing(self, temp_test_dir: Path): + """Should return None when the entry point file does not exist.""" + result = get_entry_point(str(temp_test_dir), "nonexistent") + + assert result is None + + +# ============================================================================= +# handler get_help tests +# ============================================================================= + +class TestHandlerGetHelp: + """Tests for discovery_handler.get_help().""" + + def test_returns_help_result_for_valid_entry(self, temp_test_dir: Path): + """Should return HelpResult when entry point exists and subprocess runs.""" + apps_dir = temp_test_dir / "apps" + apps_dir.mkdir(parents=True) + entry = apps_dir / "testbranch.py" + entry.write_text("# entry", encoding="utf-8") + + mock_result = MagicMock() + mock_result.stdout = b"Usage: testbranch\n\nCommands:\n run Run it\n stop Stop it\n" + mock_result.stderr = b"" + + with patch("aipass.drone.apps.handlers.discovery_handler.subprocess.run", return_value=mock_result): + result = handler_get_help(str(temp_test_dir), "testbranch") + + assert isinstance(result, HelpResult) + assert result.branch == "testbranch" + assert result.command is None + assert "run" in result.commands_found + assert "stop" in result.commands_found + + def test_returns_help_for_specific_command(self, temp_test_dir: Path): + """Should pass command argument through and set it on result.""" + apps_dir = temp_test_dir / "apps" + apps_dir.mkdir(parents=True) + (apps_dir / "testbranch.py").write_text("# entry", encoding="utf-8") + + mock_result = MagicMock() + mock_result.stdout = b"Help for the run command\n" + mock_result.stderr = b"" + + with patch("aipass.drone.apps.handlers.discovery_handler.subprocess.run", return_value=mock_result): + result = handler_get_help(str(temp_test_dir), "testbranch", command="run") + + assert result.command == "run" + assert "Help for the run command" in result.text + + def test_raises_when_entry_point_missing(self, temp_test_dir: Path): + """Should raise CommandExecutionError when no entry point exists.""" + with pytest.raises(CommandExecutionError, match="Entry point not found"): + handler_get_help(str(temp_test_dir), "nonexistent") + + def test_raises_on_timeout(self, temp_test_dir: Path): + """Should raise CommandExecutionError when subprocess times out.""" + apps_dir = temp_test_dir / "apps" + apps_dir.mkdir(parents=True) + (apps_dir / "slowbranch.py").write_text("# entry", encoding="utf-8") + + with patch( + "aipass.drone.apps.handlers.discovery_handler.subprocess.run", + side_effect=subprocess.TimeoutExpired(cmd="test", timeout=10), + ): + with pytest.raises(CommandExecutionError, match="timed out"): + handler_get_help(str(temp_test_dir), "slowbranch") + + def test_raises_on_oserror(self, temp_test_dir: Path): + """Should raise CommandExecutionError when subprocess raises OSError.""" + apps_dir = temp_test_dir / "apps" + apps_dir.mkdir(parents=True) + (apps_dir / "broken.py").write_text("# entry", encoding="utf-8") + + with patch( + "aipass.drone.apps.handlers.discovery_handler.subprocess.run", + side_effect=OSError("No such file or directory"), + ): + with pytest.raises(CommandExecutionError, match="OS error"): + handler_get_help(str(temp_test_dir), "broken") + + def test_falls_back_to_stderr(self, temp_test_dir: Path): + """When stdout is empty, should use stderr for help text.""" + apps_dir = temp_test_dir / "apps" + apps_dir.mkdir(parents=True) + (apps_dir / "testbranch.py").write_text("# entry", encoding="utf-8") + + mock_result = MagicMock() + mock_result.stdout = b"" + mock_result.stderr = b"Usage printed to stderr\n" + + with patch("aipass.drone.apps.handlers.discovery_handler.subprocess.run", return_value=mock_result): + result = handler_get_help(str(temp_test_dir), "testbranch") + + assert "Usage printed to stderr" in result.text + + +# ============================================================================= +# handler discover_modules tests +# ============================================================================= + +class TestHandlerDiscoverModules: + """Tests for discovery_handler.discover_modules().""" + + def test_uses_help_parsing_when_entry_point_exists(self, temp_test_dir: Path): + """Should prefer parsed help text when entry point exists and returns commands.""" + apps_dir = temp_test_dir / "apps" + apps_dir.mkdir(parents=True) + (apps_dir / "mybranch.py").write_text("# entry", encoding="utf-8") + + modules_dir = apps_dir / "modules" + modules_dir.mkdir() + (modules_dir / "fallback.py").write_text("", encoding="utf-8") + + mock_result = MagicMock() + mock_result.stdout = b"Commands:\n alpha Do alpha\n beta Do beta\n" + mock_result.stderr = b"" + + with patch("aipass.drone.apps.handlers.discovery_handler.subprocess.run", return_value=mock_result): + result = handler_discover_modules(str(temp_test_dir), "mybranch") + + assert "alpha" in result + assert "beta" in result + # fallback module should NOT appear when help parsing succeeded + assert "fallback" not in result + + def test_falls_back_to_scan_when_no_entry_point(self, temp_test_dir: Path): + """Should fall back to scan_modules_directory when no entry point.""" + modules_dir = temp_test_dir / "apps" / "modules" + modules_dir.mkdir(parents=True) + (modules_dir / "scanner.py").write_text("", encoding="utf-8") + (modules_dir / "parser.py").write_text("", encoding="utf-8") + + result = handler_discover_modules(str(temp_test_dir), "nonexistent") + + assert result == ["parser", "scanner"] + + def test_falls_back_to_scan_on_timeout(self, temp_test_dir: Path): + """Should fall back to scan_modules_directory when subprocess times out.""" + apps_dir = temp_test_dir / "apps" + apps_dir.mkdir(parents=True) + (apps_dir / "mybranch.py").write_text("# entry", encoding="utf-8") + + modules_dir = apps_dir / "modules" + modules_dir.mkdir() + (modules_dir / "fallback_timeout.py").write_text("", encoding="utf-8") + + with patch( + "aipass.drone.apps.handlers.discovery_handler.subprocess.run", + side_effect=subprocess.TimeoutExpired(cmd="test", timeout=10), + ): + result = handler_discover_modules(str(temp_test_dir), "mybranch") + + assert result == ["fallback_timeout"] + + def test_falls_back_to_scan_when_help_empty(self, temp_test_dir: Path): + """Should fall back to scan when help text yields no commands.""" + apps_dir = temp_test_dir / "apps" + apps_dir.mkdir(parents=True) + (apps_dir / "mybranch.py").write_text("# entry", encoding="utf-8") + + modules_dir = apps_dir / "modules" + modules_dir.mkdir() + (modules_dir / "fallback_cmd.py").write_text("", encoding="utf-8") + + mock_result = MagicMock() + mock_result.stdout = b"No commands here, just text.\n" + mock_result.stderr = b"" + + with patch("aipass.drone.apps.handlers.discovery_handler.subprocess.run", return_value=mock_result): + result = handler_discover_modules(str(temp_test_dir), "mybranch") + + assert result == ["fallback_cmd"] + + +# ============================================================================= +# orchestration layer: discovery.discover_modules tests +# ============================================================================= + +class TestOrchestrationDiscoverModules: + """Tests for discovery.discover_modules() orchestration.""" + + @patch("aipass.drone.apps.modules.discovery.resolve_branch") + @patch("aipass.drone.apps.handlers.discovery_handler.discover_modules") + def test_resolves_and_delegates(self, mock_handler_discover, mock_resolve): + """Should resolve branch path then delegate to handler.""" + from aipass.drone.apps.modules.discovery import discover_modules + + mock_resolve.return_value = "/fake/path/to/branch" + mock_handler_discover.return_value = ["cmd_a", "cmd_b"] + + result = discover_modules("@testbranch") + + mock_resolve.assert_called_once_with("@testbranch") + mock_handler_discover.assert_called_once_with("/fake/path/to/branch", "testbranch") + assert result == ["cmd_a", "cmd_b"] + + @patch("aipass.drone.apps.modules.discovery.resolve_branch") + def test_raises_on_invalid_branch(self, mock_resolve): + """Should propagate BranchNotFoundError for unknown branches.""" + from aipass.drone.apps.modules.discovery import discover_modules + + mock_resolve.side_effect = BranchNotFoundError("Branch '@nope' not found") + + with pytest.raises(BranchNotFoundError): + discover_modules("@nope") + + +# ============================================================================= +# orchestration layer: discovery.get_help tests +# ============================================================================= + +class TestOrchestrationGetHelp: + """Tests for discovery.get_help() orchestration.""" + + @patch("aipass.drone.apps.modules.discovery.resolve_branch") + @patch("aipass.drone.apps.handlers.discovery_handler.get_help") + def test_returns_help_result(self, mock_handler_help, mock_resolve): + """Should return HelpResult for a valid branch.""" + from aipass.drone.apps.modules.discovery import get_help + + mock_resolve.return_value = "/fake/branch" + mock_handler_help.return_value = HelpResult( + branch="valid", + command=None, + text="Some help text", + commands_found=["run"], + ) + + result = get_help("@valid") + + assert isinstance(result, HelpResult) + assert result.branch == "valid" + assert result.text == "Some help text" + + @patch("aipass.drone.apps.modules.discovery.resolve_branch") + def test_raises_on_invalid_branch(self, mock_resolve): + """Should propagate BranchNotFoundError for unknown branches.""" + from aipass.drone.apps.modules.discovery import get_help + + mock_resolve.side_effect = BranchNotFoundError("not found") + + with pytest.raises(BranchNotFoundError): + get_help("@nonexistent") + + +# ============================================================================= +# orchestration layer: discovery.get_system_help tests +# ============================================================================= + +class TestOrchestrationGetSystemHelp: + """Tests for discovery.get_system_help() orchestration.""" + + @patch("aipass.drone.apps.modules.discovery.list_branches") + @patch("aipass.drone.apps.handlers.discovery_handler.get_system_help") + def test_aggregates_across_branches(self, mock_sys_help, mock_list): + """Should pass active branches to handler and return results.""" + from aipass.drone.apps.modules.discovery import get_system_help + + mock_list.return_value = ["@alpha", "@beta"] + mock_sys_help.return_value = { + "alpha": HelpResult(branch="alpha", command=None, text="alpha help", commands_found=[]), + "beta": HelpResult(branch="beta", command=None, text="beta help", commands_found=[]), + } + + result = get_system_help() + + mock_list.assert_called_once_with(status="active") + assert "alpha" in result + assert "beta" in result + + +# ============================================================================= +# get_module_introspective tests +# ============================================================================= + +class TestGetModuleIntrospective: + """Tests for module_registry_handler.get_module_introspective().""" + + @patch("aipass.drone.apps.handlers.module_registry_handler._MODULE_REGISTRY", {"testmod": "fake.module.path"}) + def test_returns_introspective_output(self): + """Should call get_introspective() on the module adapter.""" + fake_mod = types.ModuleType("fake.module.path") + fake_mod.get_introspective = lambda: "Introspective info for testmod" # type: ignore[attr-defined] + + with patch("aipass.drone.apps.handlers.module_registry_handler.importlib.import_module", return_value=fake_mod): + result = get_module_introspective("testmod") + + assert result == "Introspective info for testmod" + + @patch("aipass.drone.apps.handlers.module_registry_handler._MODULE_REGISTRY", {"testmod": "fake.module.path"}) + def test_falls_back_to_help(self): + """Should fall back to get_help(None) if get_introspective is missing.""" + fake_mod = types.ModuleType("fake.module.path") + # Only attach get_help, no get_introspective + fake_mod.get_help = lambda cmd: "Help fallback text" # type: ignore[attr-defined] + + with patch("aipass.drone.apps.handlers.module_registry_handler.importlib.import_module", return_value=fake_mod): + result = get_module_introspective("testmod") + + assert result == "Help fallback text" + + def test_returns_empty_for_unknown_module(self): + """Should return empty string for unregistered module name.""" + result = get_module_introspective("totally_unknown_module_xyz") + + assert result == "" + + +# ============================================================================= +# handle_command routing tests +# ============================================================================= + +class TestHandleCommand: + """Tests for discovery.handle_command() routing.""" + + @patch("aipass.drone.apps.modules.discovery.discover_modules") + def test_modules_command_requires_arg(self, mock_discover): + """'modules' command with no args should return False.""" + from aipass.drone.apps.modules.discovery import handle_command + + result = handle_command("modules", []) + + assert result is False + mock_discover.assert_not_called() + + @patch("aipass.drone.apps.modules.discovery.discover_modules", return_value=["cmd1"]) + def test_modules_command_succeeds(self, mock_discover): + """'modules' command with a target should return True.""" + from aipass.drone.apps.modules.discovery import handle_command + + result = handle_command("modules", ["@mybranch"]) + + assert result is True + mock_discover.assert_called_once_with("@mybranch") + + def test_help_command_requires_arg(self): + """'help' command with no args should return False.""" + from aipass.drone.apps.modules.discovery import handle_command + + result = handle_command("help", []) + + assert result is False + + @patch("aipass.drone.apps.modules.discovery.get_help") + def test_help_command_with_branch(self, mock_get_help): + """'help' command with a branch target should return True.""" + from aipass.drone.apps.modules.discovery import handle_command + + mock_get_help.return_value = HelpResult( + branch="branch", command=None, text="help text", commands_found=[], + ) + + result = handle_command("help", ["@branch"]) + + assert result is True + mock_get_help.assert_called_once_with("@branch", None) + + @patch("aipass.drone.apps.modules.discovery.get_help") + def test_help_command_with_subcommand(self, mock_get_help): + """'help' command with branch and subcommand passes command through.""" + from aipass.drone.apps.modules.discovery import handle_command + + mock_get_help.return_value = HelpResult( + branch="branch", command="subcmd", text="subcmd help", commands_found=[], + ) + + result = handle_command("help", ["@branch", "subcmd"]) + + assert result is True + mock_get_help.assert_called_once_with("@branch", "subcmd") + + @patch("aipass.drone.apps.modules.discovery.get_system_help") + def test_system_command(self, mock_sys_help): + """'system' command with no args should return True.""" + from aipass.drone.apps.modules.discovery import handle_command + + mock_sys_help.return_value = { + "alpha": HelpResult(branch="alpha", command=None, text="alpha help", commands_found=[]), + } + + result = handle_command("system", []) + + assert result is True + mock_sys_help.assert_called_once() + + def test_unknown_command_returns_false(self): + """Unknown command should return False.""" + from aipass.drone.apps.modules.discovery import handle_command + + result = handle_command("nonexistent_command", []) + + assert result is False diff --git a/src/aipass/drone/tests/test_executor.py b/src/aipass/drone/tests/test_executor.py new file mode 100644 index 00000000..2fc889b8 --- /dev/null +++ b/src/aipass/drone/tests/test_executor.py @@ -0,0 +1,403 @@ +"""Tests for the subprocess executor module.""" + +import subprocess +import sys +from pathlib import Path +from unittest.mock import patch + +import pytest + +from aipass.drone.apps.handlers.exceptions import CommandExecutionError +from aipass.drone.apps.handlers.executor import execute_command + + +# --------------------------------------------------------------------------- +# 1. Captured mode — returns stdout/stderr as strings +# --------------------------------------------------------------------------- + + +class TestCapturedMode: + """Tests for captured (non-interactive) execution mode.""" + + def test_captured_stdout(self, temp_test_dir: Path): + """Captured mode returns stdout as a decoded string.""" + result = execute_command( + sys.executable, + ["-c", "print('hello world')"], + cwd=str(temp_test_dir), + ) + assert result.stdout.strip() == "hello world" + assert result.exit_code == 0 + + def test_captured_stderr(self, temp_test_dir: Path): + """Captured mode returns stderr as a decoded string.""" + result = execute_command( + sys.executable, + ["-c", "import sys; sys.stderr.write('err msg\\n')"], + cwd=str(temp_test_dir), + ) + assert "err msg" in result.stderr + assert result.exit_code == 0 + + def test_captured_both_streams(self, temp_test_dir: Path): + """Both stdout and stderr are captured simultaneously.""" + code = ( + "import sys; " + "print('out'); " + "sys.stderr.write('err\\n')" + ) + result = execute_command( + sys.executable, ["-c", code], cwd=str(temp_test_dir) + ) + assert "out" in result.stdout + assert "err" in result.stderr + + +# --------------------------------------------------------------------------- +# 2. Captured mode — timeout enforcement +# --------------------------------------------------------------------------- + + +class TestTimeout: + """Timeout enforcement in captured mode.""" + + def test_timeout_raises_command_execution_error(self, temp_test_dir: Path): + """Exceeding the timeout raises CommandExecutionError.""" + with pytest.raises(CommandExecutionError, match="timed out"): + execute_command( + sys.executable, + ["-c", "import time; time.sleep(10)"], + cwd=str(temp_test_dir), + timeout=1, + ) + + def test_timeout_chains_original_exception(self, temp_test_dir: Path): + """CommandExecutionError wraps the original TimeoutExpired.""" + with pytest.raises(CommandExecutionError) as exc_info: + execute_command( + sys.executable, + ["-c", "import time; time.sleep(10)"], + cwd=str(temp_test_dir), + timeout=1, + ) + assert isinstance(exc_info.value.__cause__, subprocess.TimeoutExpired) + + +# --------------------------------------------------------------------------- +# 3. Interactive mode — no capture +# --------------------------------------------------------------------------- + + +class TestInteractiveMode: + """Tests for interactive execution mode.""" + + def test_interactive_stdout_is_empty(self, temp_test_dir: Path): + """Interactive mode does not capture stdout.""" + result = execute_command( + sys.executable, + ["-c", "print('hello')"], + cwd=str(temp_test_dir), + interactive=True, + ) + assert result.stdout == "" + + def test_interactive_stderr_is_empty(self, temp_test_dir: Path): + """Interactive mode does not capture stderr.""" + result = execute_command( + sys.executable, + ["-c", "import sys; sys.stderr.write('err\\n')"], + cwd=str(temp_test_dir), + interactive=True, + ) + assert result.stderr == "" + + def test_interactive_exit_code_propagates(self, temp_test_dir: Path): + """Interactive mode still returns the process exit code.""" + result = execute_command( + sys.executable, + ["-c", "raise SystemExit(7)"], + cwd=str(temp_test_dir), + interactive=True, + ) + assert result.exit_code == 7 + + +# --------------------------------------------------------------------------- +# 4. Interactive mode — no timeout +# --------------------------------------------------------------------------- + + +class TestInteractiveNoTimeout: + """Interactive mode must not pass a timeout to subprocess.run.""" + + def test_no_timeout_kwarg(self, temp_test_dir: Path): + """subprocess.run is called without a timeout arg in interactive mode.""" + with patch("aipass.drone.apps.handlers.executor.subprocess.run") as mock_run: + mock_run.return_value = subprocess.CompletedProcess( + args=[], returncode=0 + ) + execute_command( + sys.executable, + ["-c", "pass"], + cwd=str(temp_test_dir), + interactive=True, + ) + call_kwargs = mock_run.call_args.kwargs + assert "timeout" not in call_kwargs + + +# --------------------------------------------------------------------------- +# 5. FileNotFoundError wraps to CommandExecutionError +# --------------------------------------------------------------------------- + + +class TestFileNotFoundWrapping: + """FileNotFoundError from a missing executable wraps correctly.""" + + def test_missing_executable_raises(self, temp_test_dir: Path): + """Non-existent executable raises CommandExecutionError.""" + with pytest.raises(CommandExecutionError, match="Executable not found"): + execute_command( + "this_executable_does_not_exist_xyz", + [], + cwd=str(temp_test_dir), + ) + + def test_missing_executable_chains_cause(self, temp_test_dir: Path): + """The original FileNotFoundError is chained.""" + with pytest.raises(CommandExecutionError) as exc_info: + execute_command( + "this_executable_does_not_exist_xyz", + [], + cwd=str(temp_test_dir), + ) + assert isinstance(exc_info.value.__cause__, FileNotFoundError) + + +# --------------------------------------------------------------------------- +# 6. OSError wraps to CommandExecutionError +# --------------------------------------------------------------------------- + + +class TestOSErrorWrapping: + """Generic OSError wraps to CommandExecutionError.""" + + def test_oserror_wraps(self, temp_test_dir: Path): + """An OSError from subprocess.run becomes CommandExecutionError.""" + with patch( + "aipass.drone.apps.handlers.executor.subprocess.run", + side_effect=OSError("mock OS failure"), + ): + with pytest.raises(CommandExecutionError, match="OS error"): + execute_command( + sys.executable, + ["-c", "pass"], + cwd=str(temp_test_dir), + ) + + def test_oserror_chains_cause(self, temp_test_dir: Path): + """The original OSError is preserved as __cause__.""" + with patch( + "aipass.drone.apps.handlers.executor.subprocess.run", + side_effect=OSError("mock OS failure"), + ): + with pytest.raises(CommandExecutionError) as exc_info: + execute_command( + sys.executable, + ["-c", "pass"], + cwd=str(temp_test_dir), + ) + assert isinstance(exc_info.value.__cause__, OSError) + + +# --------------------------------------------------------------------------- +# 7. KeyboardInterrupt in interactive mode returns exit code 130 +# --------------------------------------------------------------------------- + + +class TestKeyboardInterrupt: + """KeyboardInterrupt handling differs by mode.""" + + def test_interactive_returns_130(self, temp_test_dir: Path): + """Interactive mode catches Ctrl+C and returns exit code 130.""" + with patch( + "aipass.drone.apps.handlers.executor.subprocess.run", + side_effect=KeyboardInterrupt, + ): + result = execute_command( + sys.executable, + ["-c", "pass"], + cwd=str(temp_test_dir), + interactive=True, + ) + assert result.exit_code == 130 + assert result.stdout == "" + assert result.stderr == "" + + def test_captured_mode_reraises_keyboard_interrupt(self, temp_test_dir: Path): + """Captured mode does NOT catch KeyboardInterrupt — it propagates.""" + with patch( + "aipass.drone.apps.handlers.executor.subprocess.run", + side_effect=KeyboardInterrupt, + ): + with pytest.raises(KeyboardInterrupt): + execute_command( + sys.executable, + ["-c", "pass"], + cwd=str(temp_test_dir), + interactive=False, + ) + + +# --------------------------------------------------------------------------- +# 8. Exit codes propagate correctly +# --------------------------------------------------------------------------- + + +class TestExitCodes: + """Exit codes from the subprocess are faithfully returned.""" + + def test_exit_code_zero(self, temp_test_dir: Path): + """Successful command returns exit code 0.""" + result = execute_command( + sys.executable, + ["-c", "pass"], + cwd=str(temp_test_dir), + ) + assert result.exit_code == 0 + + def test_exit_code_one(self, temp_test_dir: Path): + """Failed command returns exit code 1.""" + result = execute_command( + sys.executable, + ["-c", "raise SystemExit(1)"], + cwd=str(temp_test_dir), + ) + assert result.exit_code == 1 + + def test_exit_code_nonzero_arbitrary(self, temp_test_dir: Path): + """Arbitrary non-zero exit code propagates.""" + result = execute_command( + sys.executable, + ["-c", "raise SystemExit(42)"], + cwd=str(temp_test_dir), + ) + assert result.exit_code == 42 + + def test_exit_code_syntax_error(self, temp_test_dir: Path): + """A Python syntax error produces non-zero exit code and stderr output.""" + result = execute_command( + sys.executable, + ["-c", "def"], + cwd=str(temp_test_dir), + ) + assert result.exit_code != 0 + assert "SyntaxError" in result.stderr + + +# --------------------------------------------------------------------------- +# 9. Custom env vars are merged with os.environ +# --------------------------------------------------------------------------- + + +class TestEnvMerging: + """Custom env dict merges with the process environment.""" + + def test_custom_env_var_visible(self, temp_test_dir: Path): + """A custom env var is available inside the subprocess.""" + result = execute_command( + sys.executable, + ["-c", "import os; print(os.environ['AIPASS_TEST_VAR'])"], + cwd=str(temp_test_dir), + env={"AIPASS_TEST_VAR": "sentinel_value_123"}, + ) + assert result.stdout.strip() == "sentinel_value_123" + + def test_existing_env_preserved(self, temp_test_dir: Path): + """Existing environment variables are still present when custom env is set.""" + import os + + expected_path = os.environ.get("PATH", "") + result = execute_command( + sys.executable, + ["-c", "import os; print(os.environ.get('PATH', ''))"], + cwd=str(temp_test_dir), + env={"AIPASS_TEST_VAR": "x"}, + ) + assert result.stdout.strip() == expected_path + + def test_no_env_uses_inherited(self, temp_test_dir: Path): + """When env=None, the subprocess inherits the parent environment.""" + import os + + expected_path = os.environ.get("PATH", "") + result = execute_command( + sys.executable, + ["-c", "import os; print(os.environ.get('PATH', ''))"], + cwd=str(temp_test_dir), + env=None, + ) + assert result.stdout.strip() == expected_path + + def test_custom_env_overrides_existing(self, temp_test_dir: Path): + """Custom env values override existing environment variables.""" + # We pick a var that definitely exists, then override it + result = execute_command( + sys.executable, + ["-c", "import os; print(os.environ['HOME'])"], + cwd=str(temp_test_dir), + env={"HOME": "/tmp/overridden"}, + ) + assert result.stdout.strip() == "/tmp/overridden" + + +# --------------------------------------------------------------------------- +# 10. shell=False is always used (security) +# --------------------------------------------------------------------------- + + +class TestShellSecurity: + """Verify shell=False is always passed to subprocess.run.""" + + def test_captured_mode_shell_false(self, temp_test_dir: Path): + """Captured mode calls subprocess.run with shell=False.""" + with patch("aipass.drone.apps.handlers.executor.subprocess.run") as mock_run: + mock_run.return_value = subprocess.CompletedProcess( + args=[], returncode=0, stdout=b"", stderr=b"" + ) + execute_command( + sys.executable, + ["-c", "pass"], + cwd=str(temp_test_dir), + ) + call_kwargs = mock_run.call_args.kwargs + assert call_kwargs["shell"] is False + + def test_interactive_mode_shell_false(self, temp_test_dir: Path): + """Interactive mode calls subprocess.run with shell=False.""" + with patch("aipass.drone.apps.handlers.executor.subprocess.run") as mock_run: + mock_run.return_value = subprocess.CompletedProcess( + args=[], returncode=0 + ) + execute_command( + sys.executable, + ["-c", "pass"], + cwd=str(temp_test_dir), + interactive=True, + ) + call_kwargs = mock_run.call_args.kwargs + assert call_kwargs["shell"] is False + + def test_shell_injection_prevented(self, temp_test_dir: Path): + """Shell metacharacters are NOT interpreted (shell=False).""" + # If shell=True were used, this would execute `echo pwned` too. + # With shell=False, the entire string is passed as one arg and + # Python will fail to parse it — confirming no shell expansion. + result = execute_command( + sys.executable, + ["-c", "import sys; print(sys.argv[1])", "hello; echo pwned"], + cwd=str(temp_test_dir), + ) + # The semicolon is treated as literal text, not a shell separator + assert result.stdout.strip() == "hello; echo pwned" + assert result.exit_code == 0 diff --git a/src/aipass/drone/tests/test_registry_handler.py b/src/aipass/drone/tests/test_registry_handler.py new file mode 100644 index 00000000..32fef4bc --- /dev/null +++ b/src/aipass/drone/tests/test_registry_handler.py @@ -0,0 +1,534 @@ +"""Tests for registry loading and credential verification. + +Covers: +- load_registry() — valid JSON, missing file, corrupt JSON, metadata parsing +- get_all_branches() — filtering, empty results +- find_registry() — directory walk-up, no registry found +- _verify_registry_credential() — ID match, ID missing, ID mismatch +""" + +import json +import os +import shutil +import tempfile +from pathlib import Path +from typing import Generator +from unittest.mock import patch + +import pytest + +from aipass.drone.apps.handlers.registry_handler import ( + _first_registry_in, + _verify_registry_credential, + find_registry, + get_all_branches, + get_registry_path, + load_registry, + reset_registry_path, + set_registry_path, +) +from aipass.drone.apps.handlers.exceptions import ( + RegistryCorruptError, + RegistryMismatchError, + RegistryNotFoundError, + RegistryPermissionError, +) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def registry_dir() -> Generator[Path, None, None]: + """Isolated temp directory for registry tests; cleaned up after.""" + d = Path(tempfile.mkdtemp(prefix="reg_test_")) + yield d + shutil.rmtree(d, ignore_errors=True) + + +@pytest.fixture(autouse=True) +def _reset_global_path(): + """Ensure the module-level _registry_path is reset between tests.""" + reset_registry_path() + yield + reset_registry_path() + + +def _write_registry(directory: Path, data: dict, name: str = "AIPASS_REGISTRY.json") -> Path: + """Helper: write a registry JSON file and return its path.""" + p = directory / name + p.write_text(json.dumps(data, indent=2), encoding="utf-8") + return p + + +def _write_passport(directory: Path, passport: dict) -> Path: + """Helper: write a .trinity/passport.json and return its path.""" + trinity = directory / ".trinity" + trinity.mkdir(parents=True, exist_ok=True) + p = trinity / "passport.json" + p.write_text(json.dumps(passport, indent=2), encoding="utf-8") + return p + + +def _minimal_registry(*, metadata_id: str | None = None, branches: list | None = None) -> dict: + """Return a minimal valid registry dict.""" + meta = {"version": "1.0.0"} + if metadata_id is not None: + meta["id"] = metadata_id + if branches is None: + branches = [ + { + "name": "Alpha", + "path": "alpha", + "type": "library", + "status": "active", + "description": "Alpha branch", + } + ] + return {"metadata": meta, "branches": branches} + + +# =================================================================== +# 1. load_registry() — valid registry +# =================================================================== + +class TestLoadRegistry: + + def test_branches_normalised_to_dict(self, registry_dir: Path): + """List-format branches are converted to a dict keyed by lowercased name.""" + reg = _minimal_registry() + _write_registry(registry_dir, reg) + set_registry_path(registry_dir / "AIPASS_REGISTRY.json") + + result = load_registry() + + assert isinstance(result, dict) + assert "branches" in result + assert "metadata" in result + assert isinstance(result["branches"], dict) + assert "alpha" in result["branches"] + assert result["branches"]["alpha"]["name"] == "alpha" + + def test_relative_paths_resolved(self, registry_dir: Path): + """Relative branch paths are resolved against registry location.""" + reg = _minimal_registry() + _write_registry(registry_dir, reg) + set_registry_path(registry_dir / "AIPASS_REGISTRY.json") + + result = load_registry() + branch_path = Path(result["branches"]["alpha"]["path"]) + + # The resolved path should be absolute + assert branch_path.is_absolute() + # It should be relative to the registry directory + assert str(branch_path).startswith(str(registry_dir)) + + def test_absolute_path_not_re_resolved(self, registry_dir: Path): + """A branch with an already-absolute path is NOT re-resolved against registry_dir.""" + abs_path = "/opt/custom/my_branch" + reg = _minimal_registry( + branches=[ + { + "name": "Absolute", + "path": abs_path, + "type": "library", + "status": "active", + } + ] + ) + _write_registry(registry_dir, reg) + set_registry_path(registry_dir / "AIPASS_REGISTRY.json") + + result = load_registry() + branch_path = result["branches"]["absolute"]["path"] + + assert branch_path == abs_path + + def test_branches_with_empty_name_skipped(self, registry_dir: Path): + """load_registry() skips branch entries with no name field.""" + reg = _minimal_registry( + branches=[ + { + "name": "Good", + "path": "good", + "type": "library", + "status": "active", + }, + { + "name": "", + "path": "empty_name", + "type": "library", + "status": "active", + }, + { + "path": "missing_name", + "type": "library", + "status": "active", + }, + ] + ) + _write_registry(registry_dir, reg) + set_registry_path(registry_dir / "AIPASS_REGISTRY.json") + + result = load_registry() + + assert len(result["branches"]) == 1 + assert "good" in result["branches"] + + def test_dict_format_branches_preserved(self, registry_dir: Path): + """If branches is already a dict, it stays a dict.""" + reg = { + "metadata": {"version": "1.0.0"}, + "branches": { + "beta": { + "name": "beta", + "path": "/tmp/beta", + "status": "active", + } + }, + } + _write_registry(registry_dir, reg) + set_registry_path(registry_dir / "AIPASS_REGISTRY.json") + + result = load_registry() + assert result["branches"]["beta"]["name"] == "beta" + + # --------------------------------------------------------------- + # 2. load_registry() — missing file + # --------------------------------------------------------------- + + def test_missing_file_raises_not_found(self, registry_dir: Path): + """load_registry() raises RegistryNotFoundError when file is absent.""" + set_registry_path(registry_dir / "NONEXISTENT_REGISTRY.json") + + with pytest.raises(RegistryNotFoundError): + load_registry() + + # --------------------------------------------------------------- + # 3. load_registry() — corrupt / invalid JSON + # --------------------------------------------------------------- + + def test_corrupt_json_raises_corrupt_error(self, registry_dir: Path): + """Invalid JSON content triggers RegistryCorruptError.""" + bad_file = registry_dir / "BAD_REGISTRY.json" + bad_file.write_text("{not valid json!!!", encoding="utf-8") + set_registry_path(bad_file) + + with pytest.raises(RegistryCorruptError): + load_registry() + + def test_non_object_json_raises_corrupt_error(self, registry_dir: Path): + """Top-level JSON that isn't a dict triggers RegistryCorruptError.""" + bad_file = registry_dir / "LIST_REGISTRY.json" + bad_file.write_text(json.dumps([1, 2, 3]), encoding="utf-8") + set_registry_path(bad_file) + + with pytest.raises(RegistryCorruptError): + load_registry() + + def test_missing_branches_key_raises_corrupt_error(self, registry_dir: Path): + """Registry dict without 'branches' triggers RegistryCorruptError.""" + no_branches = {"metadata": {"version": "1.0.0"}} + _write_registry(registry_dir, no_branches) + set_registry_path(registry_dir / "AIPASS_REGISTRY.json") + + with pytest.raises(RegistryCorruptError, match="missing 'branches'"): + load_registry() + + def test_branches_wrong_type_raises_corrupt_error(self, registry_dir: Path): + """branches field that is neither list nor dict triggers RegistryCorruptError.""" + bad = {"metadata": {"version": "1.0.0"}, "branches": "not-a-list-or-dict"} + _write_registry(registry_dir, bad) + set_registry_path(registry_dir / "AIPASS_REGISTRY.json") + + with pytest.raises(RegistryCorruptError): + load_registry() + + # --------------------------------------------------------------- + # 4. load_registry() — permission error + # --------------------------------------------------------------- + + def test_permission_error_raises_registry_permission_error(self, registry_dir: Path): + """PermissionError when reading the file triggers RegistryPermissionError.""" + reg_file = _write_registry(registry_dir, _minimal_registry()) + set_registry_path(reg_file) + + with patch("builtins.open", side_effect=PermissionError("access denied")): + with pytest.raises(RegistryPermissionError, match="Permission denied"): + load_registry() + + +# =================================================================== +# 4 & 5. get_all_branches() +# =================================================================== + +class TestGetAllBranches: + + def test_returns_list_of_branch_dicts(self, registry_dir: Path): + """get_all_branches() returns a list of dicts for active branches.""" + reg = _minimal_registry() + _write_registry(registry_dir, reg) + set_registry_path(registry_dir / "AIPASS_REGISTRY.json") + + branches = get_all_branches() + + assert isinstance(branches, list) + assert len(branches) == 1 + assert branches[0]["name"] == "alpha" + + def test_filters_by_status(self, registry_dir: Path): + """Branches with non-matching status are excluded.""" + reg = _minimal_registry( + branches=[ + {"name": "Active", "path": "a", "status": "active", "type": "lib"}, + {"name": "Archived", "path": "b", "status": "archived", "type": "lib"}, + ] + ) + _write_registry(registry_dir, reg) + set_registry_path(registry_dir / "AIPASS_REGISTRY.json") + + active = get_all_branches(status="active") + assert len(active) == 1 + assert active[0]["name"] == "active" + + archived = get_all_branches(status="archived") + assert len(archived) == 1 + assert archived[0]["name"] == "archived" + + def test_filters_by_branch_type(self, registry_dir: Path): + """branch_type filter limits results.""" + reg = _minimal_registry( + branches=[ + {"name": "Lib", "path": "l", "status": "active", "type": "library"}, + {"name": "App", "path": "a", "status": "active", "type": "app"}, + ] + ) + _write_registry(registry_dir, reg) + set_registry_path(registry_dir / "AIPASS_REGISTRY.json") + + libs = get_all_branches(branch_type="library") + assert len(libs) == 1 + assert libs[0]["type"] == "library" + + def test_returns_empty_list_when_no_registry(self, registry_dir: Path): + """When no registry file exists, get_all_branches returns [].""" + set_registry_path(registry_dir / "NONEXISTENT_REGISTRY.json") + + result = get_all_branches() + assert result == [] + + def test_returns_empty_when_no_matching_branches(self, registry_dir: Path): + """All branches filtered out yields an empty list.""" + reg = _minimal_registry( + branches=[ + {"name": "Only", "path": "o", "status": "archived", "type": "lib"}, + ] + ) + _write_registry(registry_dir, reg) + set_registry_path(registry_dir / "AIPASS_REGISTRY.json") + + result = get_all_branches(status="active") + assert result == [] + + +# =================================================================== +# 6 & 7. find_registry() / _first_registry_in() +# =================================================================== + +class TestFindRegistry: + + def test_first_registry_in_finds_file(self, registry_dir: Path): + """_first_registry_in returns path when *_REGISTRY.json exists.""" + _write_registry(registry_dir, _minimal_registry()) + + hit = _first_registry_in(registry_dir) + assert hit is not None + assert hit.name == "AIPASS_REGISTRY.json" + + def test_first_registry_in_returns_none_when_empty(self, registry_dir: Path): + """_first_registry_in returns None in an empty directory.""" + assert _first_registry_in(registry_dir) is None + + def test_first_registry_in_alphabetical(self, registry_dir: Path): + """When multiple registries exist, alphabetically first wins.""" + _write_registry(registry_dir, _minimal_registry(), name="A_REGISTRY.json") + _write_registry(registry_dir, _minimal_registry(), name="Z_REGISTRY.json") + + hit = _first_registry_in(registry_dir) + assert hit is not None + assert hit.name == "A_REGISTRY.json" + + def test_find_registry_from_child_dir(self, registry_dir: Path, monkeypatch): + """find_registry() walks up from cwd to find registry in ancestor.""" + _write_registry(registry_dir, _minimal_registry()) + child = registry_dir / "level1" / "level2" + child.mkdir(parents=True) + + monkeypatch.chdir(child) + result = find_registry() + assert result.exists() + assert result.name == "AIPASS_REGISTRY.json" + assert result.parent == registry_dir + + def test_find_registry_returns_path_when_no_registry(self, registry_dir: Path, monkeypatch): + """find_registry() returns a fallback path ending with AIPASS_REGISTRY.json when nothing found.""" + empty_child = registry_dir / "empty_sub" + empty_child.mkdir() + + monkeypatch.chdir(empty_child) + # find_registry never returns None -- it returns a conventional fallback path + result = find_registry() + assert isinstance(result, Path) + assert result.name == "AIPASS_REGISTRY.json" + + +# =================================================================== +# 8, 9, 10. _verify_registry_credential() +# =================================================================== + +class TestVerifyRegistryCredential: + + def test_passes_when_ids_match(self, registry_dir: Path, monkeypatch): + """No error when passport.citizenship.registry_id == registry.metadata.id.""" + shared_id = "reg-abc-123" + registry_data = _minimal_registry(metadata_id=shared_id) + registry_path = _write_registry(registry_dir, registry_data) + _write_passport(registry_dir, { + "citizenship": {"registry_id": shared_id} + }) + + monkeypatch.chdir(registry_dir) + # Should not raise + _verify_registry_credential(registry_path, registry_data) + + def test_passes_when_registry_id_missing(self, registry_dir: Path, monkeypatch): + """No error when registry has no metadata.id (migration period).""" + registry_data = _minimal_registry() # no metadata_id + registry_path = _write_registry(registry_dir, registry_data) + _write_passport(registry_dir, { + "citizenship": {"registry_id": "some-id"} + }) + + monkeypatch.chdir(registry_dir) + _verify_registry_credential(registry_path, registry_data) + + def test_passes_when_passport_id_missing(self, registry_dir: Path, monkeypatch): + """No error when passport has no citizenship.registry_id (migration period).""" + shared_id = "reg-xyz-789" + registry_data = _minimal_registry(metadata_id=shared_id) + registry_path = _write_registry(registry_dir, registry_data) + _write_passport(registry_dir, { + "citizenship": {} # no registry_id + }) + + monkeypatch.chdir(registry_dir) + _verify_registry_credential(registry_path, registry_data) + + def test_passes_when_no_passport_file(self, registry_dir: Path, monkeypatch): + """No error when passport.json does not exist at all.""" + shared_id = "reg-000" + registry_data = _minimal_registry(metadata_id=shared_id) + registry_path = _write_registry(registry_dir, registry_data) + # No passport written + + monkeypatch.chdir(registry_dir) + _verify_registry_credential(registry_path, registry_data) + + def test_raises_on_id_mismatch(self, registry_dir: Path, monkeypatch): + """RegistryMismatchError raised when IDs differ -- security-critical.""" + registry_data = _minimal_registry(metadata_id="registry-AAA") + registry_path = _write_registry(registry_dir, registry_data) + _write_passport(registry_dir, { + "citizenship": {"registry_id": "registry-BBB"} + }) + + monkeypatch.chdir(registry_dir) + with pytest.raises(RegistryMismatchError, match="mismatch"): + _verify_registry_credential(registry_path, registry_data) + + def test_mismatch_error_contains_both_ids(self, registry_dir: Path, monkeypatch): + """Error message includes both the passport and registry IDs for debugging.""" + reg_id = "registry-PROD" + passport_id = "registry-DEV" + registry_data = _minimal_registry(metadata_id=reg_id) + registry_path = _write_registry(registry_dir, registry_data) + _write_passport(registry_dir, { + "citizenship": {"registry_id": passport_id} + }) + + monkeypatch.chdir(registry_dir) + with pytest.raises(RegistryMismatchError) as exc_info: + _verify_registry_credential(registry_path, registry_data) + + msg = str(exc_info.value) + assert passport_id in msg + assert reg_id in msg + + def test_passes_when_metadata_key_absent(self, registry_dir: Path, monkeypatch): + """No error when the entire metadata dict is absent.""" + registry_data = {"branches": []} # no metadata at all + registry_path = registry_dir / "FAKE_REGISTRY.json" + + monkeypatch.chdir(registry_dir) + _verify_registry_credential(registry_path, registry_data) + + +# =================================================================== +# 11. Metadata parsing +# =================================================================== + +class TestMetadataParsing: + + def test_metadata_id_preserved(self, registry_dir: Path, monkeypatch): + """Registry metadata.id field is available after loading.""" + reg = _minimal_registry(metadata_id="my-unique-id") + _write_registry(registry_dir, reg) + set_registry_path(registry_dir / "AIPASS_REGISTRY.json") + + # chdir to temp so _verify_registry_credential won't find the + # real project passport (which has a different registry_id). + monkeypatch.chdir(registry_dir) + result = load_registry() + assert result["metadata"]["id"] == "my-unique-id" + + def test_metadata_version_preserved(self, registry_dir: Path): + """Registry metadata.version field survives loading.""" + reg = _minimal_registry() + _write_registry(registry_dir, reg) + set_registry_path(registry_dir / "AIPASS_REGISTRY.json") + + result = load_registry() + assert result["metadata"]["version"] == "1.0.0" + + +# =================================================================== +# Registry path management +# =================================================================== + +class TestRegistryPathManagement: + + def test_set_and_get_registry_path(self, registry_dir: Path): + """set_registry_path() overrides get_registry_path().""" + custom = registry_dir / "CUSTOM_REGISTRY.json" + set_registry_path(custom) + + assert get_registry_path() == custom + + def test_reset_clears_override(self, registry_dir: Path): + """reset_registry_path() clears any previously set override.""" + set_registry_path(registry_dir / "CUSTOM_REGISTRY.json") + reset_registry_path() + + # After reset, get_registry_path falls through to env / find + result = get_registry_path() + assert result != registry_dir / "CUSTOM_REGISTRY.json" + + def test_env_var_takes_precedence_after_reset(self, registry_dir: Path): + """AIPASS_REGISTRY env var is used when no explicit path is set.""" + env_path = str(registry_dir / "ENV_REGISTRY.json") + reset_registry_path() + + with patch.dict(os.environ, {"AIPASS_REGISTRY": env_path}): + result = get_registry_path() + assert result == Path(env_path) diff --git a/src/aipass/drone/tests/test_resolver.py b/src/aipass/drone/tests/test_resolver.py new file mode 100644 index 00000000..b1b8273b --- /dev/null +++ b/src/aipass/drone/tests/test_resolver.py @@ -0,0 +1,392 @@ +"""Tests for branch resolution (@name -> path). + +Covers resolve_branch(), list_branches(), branch_exists(), get_branch_info(), +normalize helpers, and handle_command() routing. +""" + +import json +from pathlib import Path +from typing import Any, Dict, List + +import pytest + +from aipass.drone.apps.handlers.exceptions import BranchNotFoundError +from aipass.drone.apps.handlers.registry_handler import ( + reset_registry_path, + set_registry_path, +) +from aipass.drone.apps.modules.resolver import ( + branch_exists, + get_branch_info, + handle_command, + list_branches, + normalize_branch_arg, + normalize_branch_name, + resolve_branch, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _write_registry( + registry_path: Path, + branches: List[Dict[str, Any]], + *, + metadata: Dict[str, Any] | None = None, +) -> Path: + """Write a registry file with the given branches list.""" + registry = { + "metadata": metadata or {"version": "1.0.0"}, + "branches": branches, + } + registry_path.write_text(json.dumps(registry, indent=2)) + return registry_path + + +def _make_branch( + name: str, + path: str, + *, + status: str = "active", + branch_type: str = "library", + description: str = "", +) -> Dict[str, Any]: + """Build a branch entry dict.""" + return { + "name": name, + "path": path, + "profile": branch_type, + "type": branch_type, + "description": description or f"{name} branch", + "email": f"@{name.lower()}", + "status": status, + } + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def registry_dir(temp_test_dir: Path): + """Set registry path for the duration of a test, then reset.""" + registry_path = temp_test_dir / "AIPASS_REGISTRY.json" + set_registry_path(registry_path) + yield temp_test_dir + reset_registry_path() + + +@pytest.fixture +def populated_registry(registry_dir: Path) -> Path: + """Create a registry with several branches of varying status/type.""" + branch_alpha = registry_dir / "alpha" + branch_beta = registry_dir / "beta" + branch_gamma = registry_dir / "gamma" + + registry_path = registry_dir / "AIPASS_REGISTRY.json" + _write_registry( + registry_path, + [ + _make_branch("ALPHA", str(branch_alpha), status="active", branch_type="library"), + _make_branch("BETA", str(branch_beta), status="active", branch_type="service"), + _make_branch("GAMMA", str(branch_gamma), status="archived", branch_type="library"), + ], + ) + return registry_path + + +@pytest.fixture +def single_branch_registry(registry_dir: Path, temp_test_dir: Path) -> Path: + """Registry with exactly one active branch — uses the conftest sample layout.""" + branch_path = temp_test_dir / "test_branch" + registry_path = registry_dir / "AIPASS_REGISTRY.json" + _write_registry( + registry_path, + [ + _make_branch("TEST_BRANCH", str(branch_path)), + ], + ) + return registry_path + + +@pytest.fixture +def empty_registry(registry_dir: Path) -> Path: + """Registry with an empty branches list.""" + registry_path = registry_dir / "AIPASS_REGISTRY.json" + _write_registry(registry_path, []) + return registry_path + + +# --------------------------------------------------------------------------- +# normalize helpers +# --------------------------------------------------------------------------- + +class TestNormalizeBranchName: + def test_strips_at_prefix(self): + assert normalize_branch_name("@FOO") == "FOO" + + def test_no_prefix_unchanged(self): + assert normalize_branch_name("FOO") == "FOO" + + def test_empty_string(self): + assert normalize_branch_name("") == "" + + def test_double_at_strips_one(self): + assert normalize_branch_name("@@FOO") == "@FOO" + + +class TestNormalizeBranchArg: + def test_strips_at_and_lowercases(self): + assert normalize_branch_arg("@FOO") == "foo" + + def test_no_prefix_lowercases(self): + assert normalize_branch_arg("FOO") == "foo" + + def test_multiple_at_stripped(self): + # lstrip removes all leading '@' characters + assert normalize_branch_arg("@@FOO") == "foo" + + +# --------------------------------------------------------------------------- +# resolve_branch +# --------------------------------------------------------------------------- + +class TestResolveBranch: + def test_resolve_with_at_prefix(self, populated_registry): + path = resolve_branch("@ALPHA") + assert Path(path).name == "alpha" + + def test_resolve_without_at_prefix(self, populated_registry): + path = resolve_branch("ALPHA") + assert Path(path).name == "alpha" + + def test_returns_absolute_path(self, populated_registry): + path = resolve_branch("@ALPHA") + assert Path(path).is_absolute() + + def test_case_insensitive_lookup(self, populated_registry): + """Resolver lowercases the name, registry normalizes names to lower.""" + path_upper = resolve_branch("@ALPHA") + path_lower = resolve_branch("@alpha") + path_mixed = resolve_branch("@Alpha") + assert path_upper == path_lower == path_mixed + + def test_invalid_branch_raises(self, populated_registry): + with pytest.raises(BranchNotFoundError): + resolve_branch("@NONEXISTENT") + + def test_invalid_branch_without_prefix_raises(self, populated_registry): + with pytest.raises(BranchNotFoundError): + resolve_branch("NONEXISTENT") + + def test_empty_name_raises(self, populated_registry): + with pytest.raises(BranchNotFoundError): + resolve_branch("") + + def test_at_only_raises(self, populated_registry): + with pytest.raises(BranchNotFoundError): + resolve_branch("@") + + def test_resolve_each_branch(self, populated_registry): + """Every branch in the registry should resolve.""" + for name in ("ALPHA", "BETA", "GAMMA"): + path = resolve_branch(f"@{name}") + assert Path(path).name == name.lower() + + def test_empty_registry_raises(self, empty_registry): + with pytest.raises(BranchNotFoundError): + resolve_branch("@ALPHA") + + +# --------------------------------------------------------------------------- +# branch_exists +# --------------------------------------------------------------------------- + +class TestBranchExists: + def test_exists_for_valid_branch(self, populated_registry): + assert branch_exists("@ALPHA") is True + + def test_exists_without_prefix(self, populated_registry): + assert branch_exists("ALPHA") is True + + def test_not_exists_for_unknown(self, populated_registry): + assert branch_exists("@NOPE") is False + + def test_case_insensitive(self, populated_registry): + assert branch_exists("@alpha") is True + assert branch_exists("@Alpha") is True + + def test_empty_registry(self, empty_registry): + assert branch_exists("@ALPHA") is False + + def test_archived_branch_exists(self, populated_registry): + """branch_exists uses get_branch_by_name — no status filter.""" + assert branch_exists("@GAMMA") is True + + +# --------------------------------------------------------------------------- +# get_branch_info +# --------------------------------------------------------------------------- + +class TestGetBranchInfo: + def test_contains_expected_keys(self, populated_registry): + info = get_branch_info("@ALPHA") + assert "name" in info + assert "path" in info + assert "status" in info + assert info["status"] == "active" + assert Path(info["path"]).name == "alpha" + + def test_name_is_lowercased_in_result(self, populated_registry): + info = get_branch_info("@ALPHA") + assert info["name"] == "alpha" + + def test_path_matches_resolve(self, populated_registry): + info = get_branch_info("@BETA") + path = resolve_branch("@BETA") + assert info["path"] == path + + def test_case_insensitive(self, populated_registry): + info_upper = get_branch_info("@ALPHA") + info_lower = get_branch_info("@alpha") + assert info_upper == info_lower + + def test_invalid_raises(self, populated_registry): + with pytest.raises(BranchNotFoundError): + get_branch_info("@NONEXISTENT") + + def test_without_prefix_raises_for_invalid(self, populated_registry): + with pytest.raises(BranchNotFoundError): + get_branch_info("NONEXISTENT") + + def test_empty_registry_raises(self, empty_registry): + with pytest.raises(BranchNotFoundError): + get_branch_info("@ALPHA") + + +# --------------------------------------------------------------------------- +# list_branches +# --------------------------------------------------------------------------- + +class TestListBranches: + def test_default_status_active(self, populated_registry): + """Default status='active' should exclude archived branches.""" + result = list_branches() + names_lower = [b.lower() for b in result] + assert "@alpha" in names_lower + assert "@beta" in names_lower + assert "@gamma" not in names_lower # archived + + def test_all_active_returned(self, populated_registry): + result = list_branches() + assert len(result) == 2 + + def test_filter_by_archived_status(self, populated_registry): + result = list_branches(status="archived") + assert len(result) == 1 + assert result[0].lower() == "@gamma" + + def test_filter_by_type(self, populated_registry): + result = list_branches(branch_type="service") + assert len(result) == 1 + assert result[0].lower() == "@beta" + + def test_filter_by_type_and_status(self, populated_registry): + result = list_branches(branch_type="library", status="active") + assert len(result) == 1 + assert result[0].lower() == "@alpha" + + def test_entries_have_at_prefix(self, populated_registry): + result = list_branches() + for entry in result: + assert entry.startswith("@") + + def test_empty_registry(self, empty_registry): + result = list_branches() + assert result == [] + + def test_no_matching_type(self, populated_registry): + result = list_branches(branch_type="nonexistent_type") + assert result == [] + + +# --------------------------------------------------------------------------- +# handle_command routing +# --------------------------------------------------------------------------- + +class TestHandleCommand: + def test_resolve_command_success(self, populated_registry): + assert handle_command("resolve", ["@ALPHA"]) is True + + def test_resolve_command_no_args(self, populated_registry): + assert handle_command("resolve", []) is False + + def test_exists_command_success(self, populated_registry): + assert handle_command("exists", ["@ALPHA"]) is True + + def test_exists_command_no_args(self, populated_registry): + assert handle_command("exists", []) is False + + def test_info_command_success(self, populated_registry): + assert handle_command("info", ["@ALPHA"]) is True + + def test_info_command_no_args(self, populated_registry): + assert handle_command("info", []) is False + + def test_list_command(self, populated_registry): + assert handle_command("list", []) is True + + def test_resolve_nonexistent_raises(self, populated_registry): + with pytest.raises(BranchNotFoundError): + handle_command("resolve", ["@NONEXISTENT"]) + + def test_info_nonexistent_raises(self, populated_registry): + with pytest.raises(BranchNotFoundError): + handle_command("info", ["@NONEXISTENT"]) + + def test_unknown_command(self, populated_registry): + assert handle_command("bogus", []) is False + + +# --------------------------------------------------------------------------- +# sample_registry conftest fixture +# --------------------------------------------------------------------------- + +class TestWithSampleRegistry: + """Tests using the sample_registry fixture from conftest.py.""" + + def test_resolve_via_sample(self, sample_registry): + set_registry_path(sample_registry) + try: + path = resolve_branch("@TEST_BRANCH") + assert "test_branch" in path + finally: + reset_registry_path() + + def test_branch_exists_via_sample(self, sample_registry): + set_registry_path(sample_registry) + try: + assert branch_exists("@TEST_BRANCH") is True + assert branch_exists("@MISSING") is False + finally: + reset_registry_path() + + def test_list_branches_via_sample(self, sample_registry): + set_registry_path(sample_registry) + try: + result = list_branches() + assert len(result) == 1 + assert result[0].lower() == "@test_branch" + finally: + reset_registry_path() + + def test_get_info_via_sample(self, sample_registry): + set_registry_path(sample_registry) + try: + info = get_branch_info("@TEST_BRANCH") + assert info["name"] == "test_branch" + assert info["status"] == "active" + finally: + reset_registry_path() diff --git a/src/aipass/drone/tests/test_router.py b/src/aipass/drone/tests/test_router.py new file mode 100644 index 00000000..caa7afb6 --- /dev/null +++ b/src/aipass/drone/tests/test_router.py @@ -0,0 +1,630 @@ +"""Tests for command routing — router module and router_handler.""" + +import json +import os +import sys +from pathlib import Path +from unittest.mock import patch + +import pytest + +from aipass.drone.apps.handlers.exceptions import ( + BranchNotFoundError, + CommandExecutionError, +) +from aipass.drone.apps.handlers.executor import CommandResult +from aipass.drone.apps.handlers.router_handler import ( + detect_caller_branch_name, + execute_branch_command, + find_entry_point, +) +from aipass.drone.apps.modules.router import handle_command, route_command, route_all + + +# --------------------------------------------------------------------------- +# find_entry_point +# --------------------------------------------------------------------------- + + +class TestFindEntryPoint: + """Tests for find_entry_point() in router_handler.""" + + def test_locates_existing_entry_point(self, temp_test_dir: Path): + """find_entry_point returns the correct Path when apps/{name}.py exists.""" + apps_dir = temp_test_dir / "apps" + apps_dir.mkdir() + entry_file = apps_dir / "mybranch.py" + entry_file.write_text("# entry point stub") + + result = find_entry_point(str(temp_test_dir), "mybranch") + + assert result == entry_file + assert result.exists() + + def test_raises_when_entry_point_missing(self, temp_test_dir: Path): + """find_entry_point raises CommandExecutionError if file doesn't exist.""" + with pytest.raises(CommandExecutionError, match="Entry point not found"): + find_entry_point(str(temp_test_dir), "nonexistent") + + def test_returns_path_under_apps_subdirectory(self, temp_test_dir: Path): + """Returned path is always branch_path / apps / {name}.py.""" + apps_dir = temp_test_dir / "apps" + apps_dir.mkdir() + entry_file = apps_dir / "test_branch.py" + entry_file.write_text("") + + result = find_entry_point(str(temp_test_dir), "test_branch") + + assert result.parent.name == "apps" + assert result.name == "test_branch.py" + + +# --------------------------------------------------------------------------- +# execute_branch_command +# --------------------------------------------------------------------------- + + +class TestExecuteBranchCommand: + """Tests for execute_branch_command() in router_handler.""" + + @pytest.fixture + def branch_dir(self, temp_test_dir: Path) -> Path: + """Create a fake branch directory with a valid entry point.""" + apps_dir = temp_test_dir / "apps" + apps_dir.mkdir() + entry = apps_dir / "fakebranch.py" + entry.write_text("# stub") + return temp_test_dir + + @patch("aipass.drone.apps.handlers.router_handler.execute_command") + def test_valid_command_returns_command_result( + self, mock_exec, branch_dir: Path + ): + """A valid command returns a CommandResult with correct fields.""" + mock_exec.return_value = CommandResult( + stdout="ok\n", stderr="", exit_code=0, branch="", command="" + ) + + result = execute_branch_command( + branch_path=str(branch_dir), + branch_name="fakebranch", + command="status", + ) + + assert isinstance(result, CommandResult) + assert result.exit_code == 0 + assert result.stdout == "ok\n" + assert result.branch == "fakebranch" + assert result.command == "status" + + @patch("aipass.drone.apps.handlers.router_handler.execute_command") + def test_introspection_with_command_none( + self, mock_exec, branch_dir: Path + ): + """When command=None the entry point is invoked with no command args.""" + mock_exec.return_value = CommandResult( + stdout="introspect output", stderr="", exit_code=0, branch="", command="" + ) + + result = execute_branch_command( + branch_path=str(branch_dir), + branch_name="fakebranch", + command=None, + ) + + # Args passed to execute_command should only be the relative entry point + args_list = mock_exec.call_args.kwargs["args"] + assert args_list == [str(Path("apps") / "fakebranch.py")] + assert result.command == "" + + @patch("aipass.drone.apps.handlers.router_handler.execute_command") + def test_interactive_flag_passed_through( + self, mock_exec, branch_dir: Path + ): + """interactive=True is forwarded to execute_command.""" + mock_exec.return_value = CommandResult( + stdout="", stderr="", exit_code=0, branch="", command="" + ) + + execute_branch_command( + branch_path=str(branch_dir), + branch_name="fakebranch", + command="monitor", + interactive=True, + ) + + call_kwargs = mock_exec.call_args.kwargs + assert call_kwargs.get("interactive") is True + + @patch("aipass.drone.apps.handlers.router_handler.execute_command") + def test_sets_aipass_caller_cwd_env( + self, mock_exec, branch_dir: Path + ): + """AIPASS_CALLER_CWD is set in the env dict passed to execute_command.""" + mock_exec.return_value = CommandResult( + stdout="", stderr="", exit_code=0, branch="", command="" + ) + + execute_branch_command( + branch_path=str(branch_dir), + branch_name="fakebranch", + command="test", + ) + + call_kwargs = mock_exec.call_args.kwargs + env = call_kwargs.get("env", {}) + assert "AIPASS_CALLER_CWD" in env + assert env["AIPASS_CALLER_CWD"] == str(Path.cwd()) + + @patch("aipass.drone.apps.handlers.router_handler.execute_command") + def test_timeout_propagated_to_executor( + self, mock_exec, branch_dir: Path + ): + """Timeout value is forwarded to execute_command.""" + mock_exec.return_value = CommandResult( + stdout="", stderr="", exit_code=0, branch="", command="" + ) + + execute_branch_command( + branch_path=str(branch_dir), + branch_name="fakebranch", + command="slow", + timeout=120, + ) + + call_kwargs = mock_exec.call_args.kwargs + assert call_kwargs.get("timeout") == 120 + + @patch("aipass.drone.apps.handlers.router_handler.execute_command") + def test_args_appended_to_command( + self, mock_exec, branch_dir: Path + ): + """Extra args are appended after the command in the args list.""" + mock_exec.return_value = CommandResult( + stdout="", stderr="", exit_code=0, branch="", command="" + ) + + execute_branch_command( + branch_path=str(branch_dir), + branch_name="fakebranch", + command="deploy", + args=["--force", "--env=prod"], + ) + + args_list = mock_exec.call_args.kwargs["args"] + # Verify the args list ends with the command and its arguments in exact order + assert args_list[-3:] == ["deploy", "--force", "--env=prod"] + + @patch("aipass.drone.apps.handlers.router_handler.execute_command") + def test_uses_sys_executable( + self, mock_exec, branch_dir: Path + ): + """execute_command is called with sys.executable as the executable.""" + mock_exec.return_value = CommandResult( + stdout="", stderr="", exit_code=0, branch="", command="" + ) + + execute_branch_command( + branch_path=str(branch_dir), + branch_name="fakebranch", + command="test", + ) + + call_kwargs = mock_exec.call_args.kwargs + assert call_kwargs.get("executable") == sys.executable + + def test_raises_when_entry_point_missing(self, temp_test_dir: Path): + """execute_branch_command raises CommandExecutionError if entry point missing.""" + with pytest.raises(CommandExecutionError, match="Entry point not found"): + execute_branch_command( + branch_path=str(temp_test_dir), + branch_name="does_not_exist", + command="test", + ) + + +# --------------------------------------------------------------------------- +# route_command (integration-level, mocking handler layer) +# --------------------------------------------------------------------------- + + +class TestRouteCommand: + """Tests for route_command() in the router module.""" + + @patch( + "aipass.drone.apps.modules.router.execute_branch_command" + ) + @patch("aipass.drone.apps.modules.router.resolve_branch") + def test_valid_branch_and_command(self, mock_resolve, mock_exec): + """route_command resolves target and delegates to execute_branch_command.""" + mock_resolve.return_value = "/fake/path/to/branch" + mock_exec.return_value = CommandResult( + stdout="done", stderr="", exit_code=0, branch="mybranch", command="status" + ) + + result = route_command("@mybranch", "status") + + mock_resolve.assert_called_once_with("@mybranch") + mock_exec.assert_called_once() + assert result.exit_code == 0 + assert result.stdout == "done" + + @patch("aipass.drone.apps.modules.router.resolve_branch") + def test_invalid_branch_raises_branch_not_found(self, mock_resolve): + """route_command propagates BranchNotFoundError from resolver.""" + mock_resolve.side_effect = BranchNotFoundError("Branch '@ghost' not found") + + with pytest.raises(BranchNotFoundError, match="not found"): + route_command("@ghost", "status") + + @patch( + "aipass.drone.apps.modules.router.execute_branch_command" + ) + @patch("aipass.drone.apps.modules.router.resolve_branch") + def test_timeout_forwarded(self, mock_resolve, mock_exec): + """route_command passes timeout through to execute_branch_command.""" + mock_resolve.return_value = "/fake/path" + mock_exec.return_value = CommandResult( + stdout="", stderr="", exit_code=0, branch="b", command="c" + ) + + route_command("@somebranch", "cmd", timeout=90) + + call_kwargs = mock_exec.call_args.kwargs + assert call_kwargs["timeout"] == 90 + + @patch( + "aipass.drone.apps.modules.router.execute_branch_command" + ) + @patch("aipass.drone.apps.modules.router.resolve_branch") + def test_interactive_forwarded(self, mock_resolve, mock_exec): + """route_command passes interactive flag through.""" + mock_resolve.return_value = "/fake/path" + mock_exec.return_value = CommandResult( + stdout="", stderr="", exit_code=0, branch="b", command="c" + ) + + route_command("@somebranch", "monitor", interactive=True) + + call_kwargs = mock_exec.call_args.kwargs + assert call_kwargs["interactive"] is True + + @patch( + "aipass.drone.apps.modules.router.execute_branch_command" + ) + @patch("aipass.drone.apps.modules.router.resolve_branch") + def test_introspection_no_command(self, mock_resolve, mock_exec): + """route_command with command=None triggers introspection.""" + mock_resolve.return_value = "/fake/path" + mock_exec.return_value = CommandResult( + stdout="info", stderr="", exit_code=0, branch="b", command="" + ) + + result = route_command("@somebranch", None) + + call_kwargs = mock_exec.call_args.kwargs + assert call_kwargs["command"] is None + assert result.stdout == "info" + + @patch( + "aipass.drone.apps.modules.router.execute_branch_command" + ) + @patch("aipass.drone.apps.modules.router.resolve_branch") + def test_args_forwarded(self, mock_resolve, mock_exec): + """route_command forwards args list to execute_branch_command.""" + mock_resolve.return_value = "/fake/path" + mock_exec.return_value = CommandResult( + stdout="", stderr="", exit_code=0, branch="b", command="c" + ) + + route_command("@mybranch", "deploy", args=["--env=staging"]) + + call_kwargs = mock_exec.call_args.kwargs + assert call_kwargs["args"] == ["--env=staging"] + + @patch( + "aipass.drone.apps.modules.router.execute_branch_command" + ) + @patch("aipass.drone.apps.modules.router.resolve_branch") + def test_branch_name_stripped_and_lowered(self, mock_resolve, mock_exec): + """route_command strips @ prefix and lowercases for branch_name.""" + mock_resolve.return_value = "/fake/path" + mock_exec.return_value = CommandResult( + stdout="", stderr="", exit_code=0, branch="mybranch", command="test" + ) + + route_command("@MyBranch", "test") + + call_kwargs = mock_exec.call_args.kwargs + assert call_kwargs["branch_name"] == "mybranch" + + +# --------------------------------------------------------------------------- +# route_all +# --------------------------------------------------------------------------- + + +class TestRouteAll: + """Tests for route_all() in the router module.""" + + @patch("aipass.drone.apps.modules.router.route_command") + @patch("aipass.drone.apps.modules.router.list_branches") + def test_routes_to_all_active_branches(self, mock_list, mock_route): + """route_all dispatches the command to every active branch.""" + mock_list.return_value = ["@alpha", "@beta"] + mock_route.return_value = CommandResult( + stdout="ok", stderr="", exit_code=0, branch="", command="status" + ) + + results = route_all("status") + + assert len(results) == 2 + assert "alpha" in results + assert "beta" in results + assert mock_route.call_count == 2 + + @patch("aipass.drone.apps.modules.router.route_command") + @patch("aipass.drone.apps.modules.router.list_branches") + def test_captures_failure_per_branch(self, mock_list, mock_route): + """When a branch raises, route_all records exit_code=-1 for it.""" + mock_list.return_value = ["@ok_branch", "@bad_branch"] + mock_route.side_effect = [ + CommandResult(stdout="ok", stderr="", exit_code=0, branch="ok_branch", command="cmd"), + RuntimeError("boom"), + ] + + results = route_all("cmd") + + assert results["ok_branch"].exit_code == 0 + assert results["bad_branch"].exit_code == -1 + assert "boom" in results["bad_branch"].stderr + + +# --------------------------------------------------------------------------- +# detect_caller_branch_name +# --------------------------------------------------------------------------- + + +class TestDetectCallerBranchName: + """Tests for detect_caller_branch_name() in router_handler.""" + + def test_v1_passport_branch_info(self, temp_test_dir: Path): + """Detects branch name from v1 passport format (branch_info.branch_name).""" + trinity = temp_test_dir / ".trinity" + trinity.mkdir() + passport = trinity / "passport.json" + passport.write_text(json.dumps({ + "branch_info": {"branch_name": "alpha"} + })) + + result = detect_caller_branch_name(temp_test_dir) + assert result == "alpha" + + def test_v2_passport_identity_name(self, temp_test_dir: Path): + """Detects branch name from v2 passport format (identity.name).""" + trinity = temp_test_dir / ".trinity" + trinity.mkdir() + passport = trinity / "passport.json" + passport.write_text(json.dumps({ + "identity": {"name": "beta"} + })) + + result = detect_caller_branch_name(temp_test_dir) + assert result == "beta" + + def test_v1_takes_precedence_over_v2(self, temp_test_dir: Path): + """When both v1 and v2 keys exist, v1 branch_info.branch_name wins.""" + trinity = temp_test_dir / ".trinity" + trinity.mkdir() + passport = trinity / "passport.json" + passport.write_text(json.dumps({ + "branch_info": {"branch_name": "v1name"}, + "identity": {"name": "v2name"}, + })) + + result = detect_caller_branch_name(temp_test_dir) + assert result == "v1name" + + def test_no_passport_returns_none(self, temp_test_dir: Path): + """Returns None when no .trinity/passport.json exists.""" + result = detect_caller_branch_name(temp_test_dir) + assert result is None + + def test_corrupt_passport_returns_none(self, temp_test_dir: Path): + """Returns None and doesn't crash when passport.json is invalid JSON.""" + trinity = temp_test_dir / ".trinity" + trinity.mkdir() + passport = trinity / "passport.json" + passport.write_text("{{{not valid json!!!") + + result = detect_caller_branch_name(temp_test_dir) + assert result is None + + def test_walks_up_from_subdirectory(self, temp_test_dir: Path): + """Finds passport.json in a parent directory when cwd is a subdirectory.""" + trinity = temp_test_dir / ".trinity" + trinity.mkdir() + passport = trinity / "passport.json" + passport.write_text(json.dumps({ + "branch_info": {"branch_name": "found_it"} + })) + + sub = temp_test_dir / "deep" / "nested" / "dir" + sub.mkdir(parents=True) + + result = detect_caller_branch_name(sub) + assert result == "found_it" + + +# --------------------------------------------------------------------------- +# AIPASS_CALLER_BRANCH env var +# --------------------------------------------------------------------------- + + +class TestCallerBranchEnvVar: + """Tests that execute_branch_command sets AIPASS_CALLER_BRANCH.""" + + @patch("aipass.drone.apps.handlers.router_handler.execute_command") + def test_sets_caller_branch_from_passport(self, mock_exec, temp_test_dir: Path): + """AIPASS_CALLER_BRANCH is set when passport.json exists in cwd.""" + # Set up branch entry point + apps_dir = temp_test_dir / "apps" + apps_dir.mkdir() + entry = apps_dir / "testbranch.py" + entry.write_text("# stub") + + # Set up passport in cwd + cwd_dir = temp_test_dir / "caller_cwd" + cwd_dir.mkdir() + trinity = cwd_dir / ".trinity" + trinity.mkdir() + passport = trinity / "passport.json" + passport.write_text(json.dumps({ + "branch_info": {"branch_name": "caller_branch"} + })) + + mock_exec.return_value = CommandResult( + stdout="", stderr="", exit_code=0, branch="", command="" + ) + + with patch("aipass.drone.apps.handlers.router_handler.Path") as mock_path_cls: + # Make Path.cwd() return our fake cwd + mock_path_cls.cwd.return_value = cwd_dir + # But keep Path(branch_path) / ... working for find_entry_point + mock_path_cls.side_effect = Path + + execute_branch_command( + branch_path=str(temp_test_dir), + branch_name="testbranch", + command="status", + ) + + env = mock_exec.call_args.kwargs["env"] + assert env["AIPASS_CALLER_BRANCH"] == "caller_branch" + + @patch("aipass.drone.apps.handlers.router_handler.execute_command") + def test_no_caller_branch_without_passport(self, mock_exec, temp_test_dir: Path): + """AIPASS_CALLER_BRANCH is absent when no passport.json and no env var.""" + apps_dir = temp_test_dir / "apps" + apps_dir.mkdir() + entry = apps_dir / "testbranch.py" + entry.write_text("# stub") + + # Use a cwd with no passport + cwd_dir = temp_test_dir / "empty_cwd" + cwd_dir.mkdir() + + mock_exec.return_value = CommandResult( + stdout="", stderr="", exit_code=0, branch="", command="" + ) + + with patch("aipass.drone.apps.handlers.router_handler.Path") as mock_path_cls: + mock_path_cls.cwd.return_value = cwd_dir + mock_path_cls.side_effect = Path + + with patch.dict(os.environ, {}, clear=False): + # Remove AIPASS_BRANCH_NAME if present + os.environ.pop("AIPASS_BRANCH_NAME", None) + execute_branch_command( + branch_path=str(temp_test_dir), + branch_name="testbranch", + command="status", + ) + + env = mock_exec.call_args.kwargs["env"] + assert "AIPASS_CALLER_BRANCH" not in env + + +# --------------------------------------------------------------------------- +# handle_command +# --------------------------------------------------------------------------- + + +class TestHandleCommand: + """Tests for handle_command() in the router module.""" + + def test_route_no_args_returns_false(self): + """handle_command('route', []) returns False — not enough args.""" + result = handle_command("route", []) + assert result is False + + @patch("aipass.drone.apps.modules.router.route_command") + def test_route_with_target_and_command(self, mock_route): + """handle_command('route', ['@branch', 'cmd']) delegates to route_command.""" + mock_route.return_value = CommandResult( + stdout="output", stderr="", exit_code=0, branch="branch", command="cmd" + ) + + result = handle_command("route", ["@branch", "cmd"]) + + assert result is True + mock_route.assert_called_once_with("@branch", "cmd", args=None) + + @patch("aipass.drone.apps.modules.router.route_command") + def test_route_with_extra_args(self, mock_route): + """handle_command('route', ['@b', 'cmd', '--flag']) passes extra args.""" + mock_route.return_value = CommandResult( + stdout="", stderr="", exit_code=0, branch="b", command="cmd" + ) + + result = handle_command("route", ["@b", "cmd", "--flag"]) + + assert result is True + mock_route.assert_called_once_with("@b", "cmd", args=["--flag"]) + + @patch("aipass.drone.apps.modules.router.route_command") + def test_route_nonzero_exit_returns_false(self, mock_route): + """handle_command('route', ...) returns False when exit_code != 0.""" + mock_route.return_value = CommandResult( + stdout="", stderr="err", exit_code=1, branch="b", command="cmd" + ) + + result = handle_command("route", ["@b", "cmd"]) + + assert result is False + + def test_route_all_no_args_returns_false(self): + """handle_command('route_all', []) returns False — missing command.""" + result = handle_command("route_all", []) + assert result is False + + @patch("aipass.drone.apps.modules.router.route_all") + def test_route_all_delegates(self, mock_route_all): + """handle_command('route_all', ['status']) delegates to route_all.""" + mock_route_all.return_value = { + "a": CommandResult(stdout="", stderr="", exit_code=0, branch="a", command="status"), + } + + result = handle_command("route_all", ["status"]) + + assert result is True + mock_route_all.assert_called_once_with("status", args=None) + + @patch("aipass.drone.apps.modules.router.route_all") + def test_route_all_with_extra_args(self, mock_route_all): + """handle_command('route_all', ['cmd', '--v']) passes extra args.""" + mock_route_all.return_value = { + "x": CommandResult(stdout="", stderr="", exit_code=0, branch="x", command="cmd"), + } + + result = handle_command("route_all", ["cmd", "--v"]) + + assert result is True + mock_route_all.assert_called_once_with("cmd", args=["--v"]) + + @patch("aipass.drone.apps.modules.router.route_all") + def test_route_all_partial_failure_returns_false(self, mock_route_all): + """handle_command('route_all', ...) returns False if any branch fails.""" + mock_route_all.return_value = { + "ok": CommandResult(stdout="", stderr="", exit_code=0, branch="ok", command="c"), + "bad": CommandResult(stdout="", stderr="err", exit_code=1, branch="bad", command="c"), + } + + result = handle_command("route_all", ["c"]) + + assert result is False + + def test_unknown_command_returns_false(self): + """handle_command('unknown_cmd', []) returns False.""" + result = handle_command("unknown_cmd", []) + assert result is False diff --git a/src/aipass/prax/apps/handlers/dashboard/agent_status_writer.py b/src/aipass/prax/apps/handlers/dashboard/agent_status_writer.py index fce8f13f..939cd039 100644 --- a/src/aipass/prax/apps/handlers/dashboard/agent_status_writer.py +++ b/src/aipass/prax/apps/handlers/dashboard/agent_status_writer.py @@ -9,7 +9,7 @@ """ Agent Status Dashboard Write-Through Handler -Pushes the 'agent_status' section to branch dashboards via DevPulse write_section(). +Pushes the 'agent_status' section to branch dashboards via prax write_section(). Scans dispatch lock files and /proc to detect active and stale agents, then pushes to ALL branch dashboards (agent status is system-wide info). @@ -244,7 +244,7 @@ def _write_section_to_all_branches(section_name: str, section_data: Dict, script = ( "import sys, json\n" "from pathlib import Path\n" - "from aipass.devpulse.apps.modules.dashboard import write_section\n" + "from aipass.prax.apps.modules.dashboard import write_section\n" "data = json.loads(sys.stdin.read())\n" "section_name = data['section_name']\n" "section_data = data['section_data']\n" diff --git a/src/aipass/prax/apps/handlers/dashboard/operations.py b/src/aipass/prax/apps/handlers/dashboard/operations.py index 4829515c..ca6ca4b8 100644 --- a/src/aipass/prax/apps/handlers/dashboard/operations.py +++ b/src/aipass/prax/apps/handlers/dashboard/operations.py @@ -14,13 +14,12 @@ All business logic for dashboard file operations. """ import json -import sys from datetime import datetime from pathlib import Path from typing import Dict -AIPASS_ROOT = Path.home() / "aipass_core" -sys.path.insert(0, str(AIPASS_ROOT)) +# Resolve prax root from this file's location +_PRAX_ROOT = Path(__file__).resolve().parents[3] # .../prax/ def get_dashboard_path(branch_path: Path) -> Path: @@ -112,7 +111,7 @@ def create_fresh_dashboard(branch_path: Path) -> Dict: Fresh dashboard dict with warning and all sections """ # Try loading from template file - template_file = Path.home() / "aipass_os" / "dev_central" / "devpulse" / "templates" / "DASHBOARD.template.json" + template_file = _PRAX_ROOT / "templates" / "DASHBOARD.template.json" if template_file.exists(): try: template = json.loads(template_file.read_text()) @@ -137,7 +136,6 @@ def create_fresh_dashboard(branch_path: Path) -> Dict: "ai_mail": {"managed_by": "ai_mail", "new": 0, "opened": 0, "total": 0, "last_updated": ""}, "flow": {"managed_by": "flow", "active_plans": 0, "recently_closed": [], "last_updated": ""}, "memory_bank": {"managed_by": "memory_bank", "vectors_stored": 0, "notes": {}, "last_updated": ""}, - "devpulse": {"managed_by": "devpulse", "summary": {}, "last_updated": ""}, "commons_activity": {"managed_by": "the_commons", "mentions": 0, "new_posts_since_last_visit": 0, "new_comments_since_last_visit": 0, "last_updated": ""} } } @@ -172,7 +170,6 @@ def ensure_dashboard_structure(branch_path: Path) -> Dict: "ai_mail": {"managed_by": "ai_mail", "new": 0, "opened": 0, "total": 0, "last_updated": ""}, "flow": {"managed_by": "flow", "active_plans": 0, "recently_closed": [], "last_updated": ""}, "memory_bank": {"managed_by": "memory_bank", "vectors_stored": 0, "notes": {}, "last_updated": ""}, - "devpulse": {"managed_by": "devpulse", "summary": {}, "last_updated": ""}, "commons_activity": {"managed_by": "the_commons", "mentions": 0, "new_posts_since_last_visit": 0, "new_comments_since_last_visit": 0, "last_updated": ""} } } @@ -300,7 +297,7 @@ def write_section(branch_path: Path, section_name: str, section_data: Dict) -> b - Saves back to disk Args: - branch_path: Path to branch root directory (e.g. Path("/home/aipass/aipass_os/dev_central/devpulse")) + branch_path: Path to branch root directory section_name: Section key (e.g. "ai_mail", "flow", "commons_activity") section_data: Dict of data for this section. Will be written as-is with an auto-added "last_updated" timestamp. @@ -309,8 +306,8 @@ def write_section(branch_path: Path, section_name: str, section_data: Dict) -> b True if saved successfully, False on any error Example: - >>> from aipass_os.dev_central.devpulse.apps.modules.dashboard import write_section - >>> write_section(Path("/home/aipass/aipass_os/flow"), "flow", {"active_plans": 2}) + >>> from aipass.prax.apps.modules.dashboard import write_section + >>> write_section(branch_path, "flow", {"active_plans": 2}) True """ try: diff --git a/src/aipass/prax/apps/handlers/dashboard/refresh.py b/src/aipass/prax/apps/handlers/dashboard/refresh.py index e1b450d9..f2731abc 100644 --- a/src/aipass/prax/apps/handlers/dashboard/refresh.py +++ b/src/aipass/prax/apps/handlers/dashboard/refresh.py @@ -25,11 +25,20 @@ from .operations import create_fresh_dashboard, save_dashboard from ..central.reader import read_all_centrals # Sections managed by the refresh path — everything else is write-through only -REFRESH_MANAGED_SECTIONS = {"ai_mail", "flow", "memory_bank", "devpulse", "commons_activity"} +REFRESH_MANAGED_SECTIONS = {"ai_mail", "flow", "memory_bank", "commons_activity"} + + +def _find_repo_root() -> Path: + """Walk up from this file to find the repo root (contains AIPASS_REGISTRY.json).""" + current = Path(__file__).resolve().parent + for parent in [current] + list(current.parents): + if (parent / "AIPASS_REGISTRY.json").exists(): + return parent + return Path.cwd() + # Infrastructure -AIPASS_ROOT = Path.home() -BRANCH_REGISTRY = AIPASS_ROOT / "BRANCH_REGISTRY.json" +BRANCH_REGISTRY = _find_repo_root() / "AIPASS_REGISTRY.json" def _load_branch_paths() -> List[Path]: @@ -45,6 +54,7 @@ def _load_branch_paths() -> List[Path]: if not BRANCH_REGISTRY.exists(): raise FileNotFoundError(f"Branch registry not found: {BRANCH_REGISTRY}") + repo_root = _find_repo_root() data = json.loads(BRANCH_REGISTRY.read_text()) branches = data.get("branches", []) @@ -52,7 +62,8 @@ def _load_branch_paths() -> List[Path]: for branch in branches: path_str = branch.get("path") if path_str: - path = Path(path_str) + raw = Path(path_str) + path = raw if raw.is_absolute() else repo_root / raw if path.exists(): paths.append(path) @@ -139,29 +150,6 @@ def _extract_memory_bank_section(centrals: Dict, branch_path: Path) -> Dict: } -def _extract_devpulse_section(centrals: Dict, branch_name: str) -> Dict: - """Extract devpulse section from DEVPULSE.central.json, including dplan_counts.""" - dp_data = centrals.get("devpulse") - if not dp_data: - return {"managed_by": "devpulse", "summary": {}, "dplan_counts": {}, "recent_activity": ""} - - summaries = dp_data.get("branch_summaries", {}) - branch_summary = summaries.get(branch_name, {}) - - # Include dplan_counts from central dplan_summary (Phase 4 enrichment) - dplan_summary = dp_data.get("dplan_summary", {}) - dplan_counts = dplan_summary.get("dplan_counts", {}) - recent_activity = dplan_summary.get("recent_activity", "") - - return { - "managed_by": "devpulse", - "summary": branch_summary, - "dplan_counts": dplan_counts, - "recent_activity": recent_activity, - "last_updated": dp_data.get("last_updated", datetime.now().isoformat()) - } - - def _extract_commons_section(centrals: Dict, branch_name: str) -> Optional[Dict]: """ Extract commons_activity section from COMMONS.central.json. @@ -277,7 +265,6 @@ def refresh_all_dashboards() -> Dict: dashboard["sections"]["ai_mail"] = _extract_ai_mail_section(centrals, branch_name) dashboard["sections"]["flow"] = _extract_flow_section(centrals, branch_name) dashboard["sections"]["memory_bank"] = _extract_memory_bank_section(centrals, branch_path) - dashboard["sections"]["devpulse"] = _extract_devpulse_section(centrals, branch_name) # Commons: preserve existing write-through data if no central file commons_section = _extract_commons_section(centrals, branch_name) @@ -351,7 +338,6 @@ def refresh_single_dashboard(branch_path: Path) -> Dict: dashboard["sections"]["ai_mail"] = _extract_ai_mail_section(centrals, branch_name) dashboard["sections"]["flow"] = _extract_flow_section(centrals, branch_name) dashboard["sections"]["memory_bank"] = _extract_memory_bank_section(centrals, branch_path) - dashboard["sections"]["devpulse"] = _extract_devpulse_section(centrals, branch_name) # Commons: preserve existing write-through data if no central file commons_section = _extract_commons_section(centrals, branch_name) diff --git a/src/aipass/prax/apps/handlers/dashboard/status.py b/src/aipass/prax/apps/handlers/dashboard/status.py index 457c347f..8a6c1965 100644 --- a/src/aipass/prax/apps/handlers/dashboard/status.py +++ b/src/aipass/prax/apps/handlers/dashboard/status.py @@ -17,8 +17,17 @@ import json from pathlib import Path from typing import Dict, List -AIPASS_ROOT = Path.home() -BRANCH_REGISTRY = AIPASS_ROOT / "BRANCH_REGISTRY.json" + +def _find_repo_root() -> Path: + """Walk up from this file to find the repo root (contains AIPASS_REGISTRY.json).""" + current = Path(__file__).resolve().parent + for parent in [current] + list(current.parents): + if (parent / "AIPASS_REGISTRY.json").exists(): + return parent + return Path.cwd() + + +BRANCH_REGISTRY = _find_repo_root() / "AIPASS_REGISTRY.json" def calculate_quick_status(sections: Dict) -> Dict: @@ -81,13 +90,18 @@ def get_branch_paths() -> List[Path]: if not BRANCH_REGISTRY.exists(): raise FileNotFoundError(f"Branch registry not found: {BRANCH_REGISTRY}") + repo_root = _find_repo_root() data = json.loads(BRANCH_REGISTRY.read_text()) - return [Path(b.get("path", "")) for b in data.get("branches", [])] + paths = [] + for b in data.get("branches", []): + raw = Path(b.get("path", "")) + paths.append(raw if raw.is_absolute() else repo_root / raw) + return paths def resolve_branch_path(branch_ref: str) -> Path: """ - Resolve @branch reference to filesystem path via BRANCH_REGISTRY.json. + Resolve @branch reference to filesystem path via AIPASS_REGISTRY.json. Handler-layer function: performs file I/O to read registry and resolve branch name to its directory path. @@ -104,12 +118,14 @@ def resolve_branch_path(branch_ref: str) -> Path: name = branch_ref.lstrip("@").upper() if not BRANCH_REGISTRY.exists(): - raise FileNotFoundError("BRANCH_REGISTRY.json not found") + raise FileNotFoundError("AIPASS_REGISTRY.json not found") + repo_root = _find_repo_root() data = json.loads(BRANCH_REGISTRY.read_text()) for branch in data.get("branches", []): if branch.get("name", "").upper() == name: - path = Path(branch["path"]) + raw = Path(branch["path"]) + path = raw if raw.is_absolute() else repo_root / raw if path.exists(): return path raise FileNotFoundError(f"Branch path does not exist: {path}") diff --git a/src/aipass/prax/apps/handlers/dashboard/template_differ.py b/src/aipass/prax/apps/handlers/dashboard/template_differ.py index e74d16b5..e35f817c 100644 --- a/src/aipass/prax/apps/handlers/dashboard/template_differ.py +++ b/src/aipass/prax/apps/handlers/dashboard/template_differ.py @@ -23,32 +23,42 @@ Independence: """ import json -import sys from pathlib import Path -from typing import Dict, Any, List, Optional +from typing import Dict, Any, Optional + +# ============================================================================= +# PATH RESOLUTION +# ============================================================================= + +_PRAX_ROOT = Path(__file__).resolve().parents[3] # .../prax/ + + +def _find_repo_root() -> Path: + """Walk up from this file to find the repo root (contains AIPASS_REGISTRY.json).""" + current = Path(__file__).resolve().parent + for parent in [current] + list(current.parents): + if (parent / "AIPASS_REGISTRY.json").exists(): + return parent + return Path.cwd() -# Infrastructure setup -AIPASS_ROOT = Path.home() -sys.path.insert(0, str(AIPASS_ROOT)) # ============================================================================= # CONSTANTS # ============================================================================= -DEVPULSE_ROOT = AIPASS_ROOT / "aipass_os" / "dev_central" / "devpulse" -TEMPLATE_DIR = DEVPULSE_ROOT / "templates" +TEMPLATE_DIR = _PRAX_ROOT / "templates" TEMPLATE_FILE = TEMPLATE_DIR / "DASHBOARD.template.json" -BRANCH_REGISTRY = AIPASS_ROOT / "BRANCH_REGISTRY.json" +BRANCH_REGISTRY = _find_repo_root() / "AIPASS_REGISTRY.json" # Deprecated sections that should be flagged for removal -DEPRECATED_SECTIONS = ["bulletin_board"] +DEPRECATED_SECTIONS = ["bulletin_board", "devpulse"] # Deprecated quick_status keys that should be flagged DEPRECATED_QUICK_STATUS_KEYS = ["pending_bulletins"] # Required sections (from template) REQUIRED_SECTIONS = [ - "ai_mail", "flow", "memory_bank", "devpulse", "commons_activity" + "ai_mail", "flow", "memory_bank", "commons_activity" ] @@ -107,7 +117,6 @@ def _diff_branch(branch_name: str, branch_path: Path, template: dict) -> Dict[st return result # Check for missing required sections - template_sections = template.get("sections", {}) for section_name in REQUIRED_SECTIONS: if section_name not in sections: result["additions"].append(f"{section_name} section") @@ -202,9 +211,12 @@ def diff_dashboard_template(branch_name: Optional[str] = None) -> Dict[str, Any] "summary": {} } + repo_root = _find_repo_root() + for branch in branches: bname = branch.get("name", "UNKNOWN").upper() - bpath = Path(branch.get("path", "")) + raw_path = Path(branch.get("path", "")) + bpath = raw_path if raw_path.is_absolute() else repo_root / raw_path if not bpath.exists(): branch_diff = { diff --git a/src/aipass/prax/apps/handlers/dashboard/template_pusher.py b/src/aipass/prax/apps/handlers/dashboard/template_pusher.py index 52235637..e9690825 100644 --- a/src/aipass/prax/apps/handlers/dashboard/template_pusher.py +++ b/src/aipass/prax/apps/handlers/dashboard/template_pusher.py @@ -25,27 +25,37 @@ Independence: import json import copy -import sys from pathlib import Path from datetime import datetime -from typing import Dict, Any, List, Optional +from typing import Dict, Any, List + +# ============================================================================= +# PATH RESOLUTION +# ============================================================================= + +_PRAX_ROOT = Path(__file__).resolve().parents[3] # .../prax/ + + +def _find_repo_root() -> Path: + """Walk up from this file to find the repo root (contains AIPASS_REGISTRY.json).""" + current = Path(__file__).resolve().parent + for parent in [current] + list(current.parents): + if (parent / "AIPASS_REGISTRY.json").exists(): + return parent + return Path.cwd() -# Infrastructure setup -AIPASS_ROOT = Path.home() -sys.path.insert(0, str(AIPASS_ROOT)) # ============================================================================= # CONSTANTS # ============================================================================= -DEVPULSE_ROOT = AIPASS_ROOT / "aipass_os" / "dev_central" / "devpulse" -TEMPLATE_DIR = DEVPULSE_ROOT / "templates" +TEMPLATE_DIR = _PRAX_ROOT / "templates" TEMPLATE_FILE = TEMPLATE_DIR / "DASHBOARD.template.json" VERSION_FILE = TEMPLATE_DIR / ".dashboard_version.json" -BRANCH_REGISTRY = AIPASS_ROOT / "BRANCH_REGISTRY.json" +BRANCH_REGISTRY = _find_repo_root() / "AIPASS_REGISTRY.json" # Deprecated sections to REMOVE during push -DEPRECATED_SECTIONS = ["bulletin_board"] +DEPRECATED_SECTIONS = ["bulletin_board", "devpulse"] # Deprecated quick_status keys to REMOVE during push DEPRECATED_QUICK_STATUS_KEYS = ["pending_bulletins"] @@ -71,11 +81,6 @@ REQUIRED_SECTIONS = { "notes": {}, "last_updated": "" }, - "devpulse": { - "managed_by": "devpulse", - "summary": {}, - "last_updated": "" - }, "commons_activity": { "managed_by": "the_commons", "mentions": 0, @@ -229,9 +234,12 @@ def push_dashboard_template(dry_run: bool = False) -> Dict[str, Any]: branches = [b for b in registry.get("branches", []) if b.get("status") == "active"] branches_updated_list: List[str] = [] + repo_root = _find_repo_root() + for branch in branches: branch_name = branch.get("name", "UNKNOWN").upper() - branch_path = Path(branch.get("path", "")) + raw_path = Path(branch.get("path", "")) + branch_path = raw_path if raw_path.is_absolute() else repo_root / raw_path result["branches_scanned"] += 1 if not branch_path.exists(): diff --git a/src/aipass/prax/apps/modules/dashboard.py b/src/aipass/prax/apps/modules/dashboard.py index 3d9ef1f6..4cce4e09 100644 --- a/src/aipass/prax/apps/modules/dashboard.py +++ b/src/aipass/prax/apps/modules/dashboard.py @@ -79,11 +79,6 @@ DASHBOARD_TEMPLATE = { "notes": {}, "last_updated": "" }, - "devpulse": { - "managed_by": "devpulse", - "summary": {}, - "last_updated": "" - }, "commons_activity": { "managed_by": "the_commons", "mentions": 0, @@ -115,7 +110,7 @@ def update_section( Update a specific section in branch dashboard (legacy wrapper). For new integrations, prefer write_section() from - aipass_os.dev_central.devpulse.apps.handlers.dashboard.operations + aipass.prax.apps.handlers.dashboard.operations which is self-contained and dependency-free. Args: @@ -170,19 +165,19 @@ def print_help(): console.print(" template-status - Show template version and push info") console.print() console.print("[yellow]USAGE:[/yellow]") - console.print(" drone @devpulse dashboard status") - console.print(" drone @devpulse dashboard template") - console.print(" drone @devpulse dashboard refresh # refresh current branch") - console.print(" drone @devpulse dashboard refresh @flow # refresh specific branch") - console.print(" drone @devpulse dashboard refresh --all # refresh all branches") - console.print(" drone @devpulse dashboard push-template # push to all branches") - console.print(" drone @devpulse dashboard push-template --dry-run # preview changes") - console.print(" drone @devpulse dashboard diff-template # diff all branches") - console.print(" drone @devpulse dashboard diff-template --branch FLOW # diff single branch") - console.print(" drone @devpulse dashboard template-status # version info") + console.print(" drone @prax dashboard status") + console.print(" drone @prax dashboard template") + console.print(" drone @prax dashboard refresh # refresh current branch") + console.print(" drone @prax dashboard refresh @flow # refresh specific branch") + console.print(" drone @prax dashboard refresh --all # refresh all branches") + console.print(" drone @prax dashboard push-template # push to all branches") + console.print(" drone @prax dashboard push-template --dry-run # preview changes") + console.print(" drone @prax dashboard diff-template # diff all branches") + console.print(" drone @prax dashboard diff-template --branch FLOW # diff single branch") + console.print(" drone @prax dashboard template-status # version info") console.print() console.print("[yellow]PROGRAMMATIC (write-through API):[/yellow]") - console.print(" from aipass_os.dev_central.devpulse.apps.modules.dashboard import write_section") + console.print(" from aipass.prax.apps.modules.dashboard import write_section") console.print(" write_section(branch_path, 'ai_mail', {'new': 3, 'total': 5})") console.print() diff --git a/src/aipass/prax/templates/.dashboard_version.json b/src/aipass/prax/templates/.dashboard_version.json index a410ce26..4dc7a02e 100644 --- a/src/aipass/prax/templates/.dashboard_version.json +++ b/src/aipass/prax/templates/.dashboard_version.json @@ -10,35 +10,22 @@ "description": "Branch dashboard template - v3 schema with write-through sections" } }, - "last_push": "2026-02-25 18:23:16", + "last_push": "2026-03-14 21:30:50", "last_push_branches": [ - ".VSCODE", "AI_MAIL", "API", - "ASSISTANT", - "BACKUP_SYSTEM", + "BACKUP", "CLI", - "CORTEX", + "COMMONS", + "DAEMON", "DEVPULSE", - "DEV_CENTRAL", "DRONE", "FLOW", - "GIT_MANAGER", - "GROWTH", - "MCP_SERVERS", - "MEMORY_BANK", - "NEXUS", - "PERMISSIONS", + "MEMORY", "PRAX", - "PROJECTS", - "SEED", - "SPEAKEASY", - "TEAM_1", - "TEAM_2", - "TEAM_3", - "TEST", - "THE_COMMONS", - "TRIGGER", - "VERA" + "SEEDGO", + "SKILLS", + "SPAWN", + "TRIGGER" ] } diff --git a/src/aipass/prax/templates/DASHBOARD.template.json b/src/aipass/prax/templates/DASHBOARD.template.json index a2227630..416680fc 100644 --- a/src/aipass/prax/templates/DASHBOARD.template.json +++ b/src/aipass/prax/templates/DASHBOARD.template.json @@ -30,11 +30,6 @@ "notes": {}, "last_updated": "" }, - "devpulse": { - "managed_by": "devpulse", - "summary": {}, - "last_updated": "" - }, "commons_activity": { "managed_by": "the_commons", "mentions": 0,