From 5f514604f7b27f7249d55c39bafec4afe6929c6d Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Mon, 15 Jun 2026 09:30:58 -0700 Subject: [PATCH] feat(skills): wire telegram handler.py run() -> bot_operations (FPLAN-0277 live bring-up) Replace scaffold stub with dispatch table routing all 6 actions (start/stop/ status/create/delete/notify) to bot_operations, bot_factory, notifier. Lazy imports per action, arg validation, graceful error returns. +28 routing tests (test_handler_routing.py). Verified live: status -> real registry query. TG 445/452 (7 telethon-absent), skills 252/252 no regressions. seedgo 25/27 (skill-folder FP + required lazy imports). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../skills/.aipass/skills/telegram/handler.py | 142 +++++++++++++- .../telegram/tests/test_handler_routing.py | 185 ++++++++++++++++++ 2 files changed, 318 insertions(+), 9 deletions(-) create mode 100644 src/aipass/skills/.aipass/skills/telegram/tests/test_handler_routing.py diff --git a/src/aipass/skills/.aipass/skills/telegram/handler.py b/src/aipass/skills/.aipass/skills/telegram/handler.py index 2db50747..c95993eb 100644 --- a/src/aipass/skills/.aipass/skills/telegram/handler.py +++ b/src/aipass/skills/.aipass/skills/telegram/handler.py @@ -1,7 +1,7 @@ # =================== AIPass ==================== # Name: handler.py # Description: Telegram skill entry point — routes actions to multi-bot framework -# Version: 1.0.0 +# Version: 1.1.0 # Created: 2026-06-15 # Modified: 2026-06-15 # ============================================= @@ -12,21 +12,145 @@ telegram — Full 3-layer skill handler. Routes drone @skills run telegram to the multi-bot framework. """ +import sys +from pathlib import Path + +from aipass.prax import logger + +_skill_root = Path(__file__).resolve().parent +if str(_skill_root) not in sys.path: + sys.path.insert(0, str(_skill_root)) + +_ACTIONS = { + "start", + "stop", + "status", + "create", + "delete", + "notify", +} + + +def _ok(output: str) -> dict: + return {"success": True, "output": output, "error": None} + + +def _err(msg: str) -> dict: + return {"success": False, "output": "", "error": msg} + + +def _cmd_start(args: list) -> dict: + if not args: + return _err("start requires a bot_id: drone @skills run telegram start ") + from apps.handlers.bot_operations import start_bot + + bot_id = args[0] + exit_code = start_bot(bot_id) + if exit_code is None: + return _err(f"Failed to load config for bot '{bot_id}'") + return _ok(f"Bot '{bot_id}' exited with code {exit_code}") + + +def _cmd_stop(args: list) -> dict: + if not args: + return _err("stop requires a bot_id: drone @skills run telegram stop ") + from apps.handlers.bot_operations import stop_bot + + success, message = stop_bot(args[0]) + if success: + return _ok(message) + return _err(message) + + +def _cmd_status(args: list) -> dict: + from apps.handlers.bot_operations import ( + format_bot_details, + format_bot_table, + get_status, + ) + + bot_id = args[0] if args else None + bots = get_status(bot_id) + if not bots: + msg = f"No bot found with id '{bot_id}'" if bot_id else "No bots registered" + return _ok(msg) + if bot_id: + return _ok("\n".join(format_bot_details(bots[0]))) + return _ok("\n".join(format_bot_table(bots))) + + +def _cmd_create(args: list) -> dict: + from apps.handlers.bot_factory import create_bot + from apps.handlers.bot_operations import parse_create_args + + parsed = parse_create_args(args) + if not parsed: + return _err("create requires: [--branch name] [--work-dir path]") + result = create_bot( + bot_id=parsed["bot_id"], + bot_token=parsed["bot_token"], + branch_name=parsed.get("branch_name"), + work_dir=parsed.get("work_dir"), + ) + if result: + return _ok(f"Bot '{parsed['bot_id']}' created successfully") + return _err(f"Failed to create bot '{parsed['bot_id']}'") + + +def _cmd_delete(args: list) -> dict: + if not args: + return _err("delete requires a bot_id: drone @skills run telegram delete ") + from apps.handlers.bot_factory import delete_bot + + success = delete_bot(args[0]) + if success: + return _ok(f"Bot '{args[0]}' deleted") + return _err(f"Failed to delete bot '{args[0]}'") + + +def _cmd_notify(args: list) -> dict: + if not args: + return _err('notify requires a message: drone @skills run telegram notify "message"') + from apps.handlers.notifier import send_telegram_notification + + message = " ".join(args) + success = send_telegram_notification(message) + if success: + return _ok(f"Notification sent: {message}") + return _err("Failed to send notification") + + +_DISPATCH = { + "start": _cmd_start, + "stop": _cmd_stop, + "status": _cmd_status, + "create": _cmd_create, + "delete": _cmd_delete, + "notify": _cmd_notify, +} + def run(action: str, args: list, config: dict) -> dict: """ - Execute the skill. + Route a skill action to the telegram multi-bot framework. Args: - action: The action to perform - args: Command arguments + action: The action to perform (start, stop, status, create, delete, notify) + args: Command arguments after the action config: Skill configuration from SKILL.md Returns: dict with keys: success (bool), output (str), error (str|None) """ - return { - "success": True, - "output": f"telegram executed action: {action}", - "error": None, - } + if not action: + return _err(f"No action specified. Available: {', '.join(sorted(_ACTIONS))}") + + handler = _DISPATCH.get(action) + if not handler: + return _err(f"Unknown action '{action}'. Available: {', '.join(sorted(_ACTIONS))}") + + try: + return handler(args or []) + except Exception as e: + logger.error("telegram skill action '%s' failed: %s", action, e) + return _err(f"{action} failed: {e}") diff --git a/src/aipass/skills/.aipass/skills/telegram/tests/test_handler_routing.py b/src/aipass/skills/.aipass/skills/telegram/tests/test_handler_routing.py new file mode 100644 index 00000000..46d28087 --- /dev/null +++ b/src/aipass/skills/.aipass/skills/telegram/tests/test_handler_routing.py @@ -0,0 +1,185 @@ +"""Tests for handler.py action routing.""" + +import sys +from pathlib import Path +from unittest.mock import patch + + +# Ensure skill root is on path so handler can import apps.handlers.* +_skill_root = Path(__file__).resolve().parents[1] +if str(_skill_root) not in sys.path: + sys.path.insert(0, str(_skill_root)) + +from handler import run, _ok, _err, _ACTIONS, _DISPATCH + + +class TestHelpers: + def test_ok_returns_success_dict(self): + result = _ok("hello") + assert result == {"success": True, "output": "hello", "error": None} + + def test_err_returns_failure_dict(self): + result = _err("bad thing") + assert result == {"success": False, "output": "", "error": "bad thing"} + + +class TestRunRouting: + def test_empty_action_returns_error(self): + result = run("", [], {}) + assert result["success"] is False + assert "No action specified" in result["error"] + + def test_unknown_action_returns_error(self): + result = run("bogus", [], {}) + assert result["success"] is False + assert "Unknown action 'bogus'" in result["error"] + assert "create" in result["error"] + + def test_all_actions_have_dispatch_entry(self): + for action in _ACTIONS: + assert action in _DISPATCH + + def test_dispatch_covers_all_actions(self): + assert set(_DISPATCH.keys()) == _ACTIONS + + +class TestStartAction: + def test_start_no_args_returns_error(self): + result = run("start", [], {}) + assert result["success"] is False + assert "bot_id" in result["error"] + + @patch("apps.handlers.bot_operations.start_bot", return_value=0) + def test_start_routes_to_start_bot(self, mock_start): + result = run("start", ["base"], {}) + mock_start.assert_called_once_with("base") + assert result["success"] is True + assert "base" in result["output"] + + @patch("apps.handlers.bot_operations.start_bot", return_value=None) + def test_start_config_fail_returns_error(self, mock_start): + result = run("start", ["missing"], {}) + assert result["success"] is False + assert "missing" in result["error"] + + +class TestStopAction: + def test_stop_no_args_returns_error(self): + result = run("stop", [], {}) + assert result["success"] is False + assert "bot_id" in result["error"] + + @patch("apps.handlers.bot_operations.stop_bot", return_value=(True, "Stopped telegram-bot@base")) + def test_stop_success(self, mock_stop): + result = run("stop", ["base"], {}) + mock_stop.assert_called_once_with("base") + assert result["success"] is True + + @patch("apps.handlers.bot_operations.stop_bot", return_value=(False, "Service not found")) + def test_stop_failure(self, mock_stop): + result = run("stop", ["base"], {}) + assert result["success"] is False + + +class TestStatusAction: + @patch("apps.handlers.bot_operations.get_status", return_value=[]) + def test_status_no_bots(self, mock_status): + result = run("status", [], {}) + assert result["success"] is True + assert "No bots registered" in result["output"] + + @patch("apps.handlers.bot_operations.get_status", return_value=[]) + def test_status_specific_bot_not_found(self, mock_status): + result = run("status", ["missing"], {}) + assert result["success"] is True + assert "missing" in result["output"] + + @patch("apps.handlers.bot_operations.format_bot_details", return_value=["Bot ID: base", "Status: running"]) + @patch("apps.handlers.bot_operations.get_status", return_value=[{"bot_id": "base", "status": "running"}]) + def test_status_specific_bot_found(self, mock_status, mock_format): + result = run("status", ["base"], {}) + assert result["success"] is True + assert "Bot ID: base" in result["output"] + + @patch( + "apps.handlers.bot_operations.format_bot_table", + return_value=["Bot ID Branch Status", "base - running"], + ) + @patch("apps.handlers.bot_operations.get_status", return_value=[{"bot_id": "base"}, {"bot_id": "dev"}]) + def test_status_all_bots(self, mock_status, mock_table): + result = run("status", [], {}) + assert result["success"] is True + assert "Bot ID" in result["output"] + + +class TestCreateAction: + def test_create_no_args_returns_error(self): + result = run("create", [], {}) + assert result["success"] is False + assert "requires" in result["error"] + + def test_create_missing_token_returns_error(self): + result = run("create", ["mybot"], {}) + assert result["success"] is False + + @patch("apps.handlers.bot_factory.create_bot", return_value={"bot_id": "mybot"}) + def test_create_success(self, mock_create): + result = run("create", ["mybot", "123:ABC"], {}) + mock_create.assert_called_once_with(bot_id="mybot", bot_token="123:ABC", branch_name=None, work_dir=None) + assert result["success"] is True + assert "mybot" in result["output"] + + @patch("apps.handlers.bot_factory.create_bot", return_value=None) + def test_create_failure(self, mock_create): + result = run("create", ["mybot", "123:ABC"], {}) + assert result["success"] is False + + @patch("apps.handlers.bot_factory.create_bot", return_value={"bot_id": "mybot"}) + def test_create_with_branch_flag(self, mock_create): + result = run("create", ["mybot", "123:ABC", "--branch", "dev"], {}) + mock_create.assert_called_once_with(bot_id="mybot", bot_token="123:ABC", branch_name="dev", work_dir=None) + assert result["success"] is True + + +class TestDeleteAction: + def test_delete_no_args_returns_error(self): + result = run("delete", [], {}) + assert result["success"] is False + assert "bot_id" in result["error"] + + @patch("apps.handlers.bot_factory.delete_bot", return_value=True) + def test_delete_success(self, mock_delete): + result = run("delete", ["base"], {}) + mock_delete.assert_called_once_with("base") + assert result["success"] is True + + @patch("apps.handlers.bot_factory.delete_bot", return_value=False) + def test_delete_failure(self, mock_delete): + result = run("delete", ["base"], {}) + assert result["success"] is False + + +class TestNotifyAction: + def test_notify_no_args_returns_error(self): + result = run("notify", [], {}) + assert result["success"] is False + assert "message" in result["error"] + + @patch("apps.handlers.notifier.send_telegram_notification", return_value=True) + def test_notify_success(self, mock_notify): + result = run("notify", ["hello", "world"], {}) + mock_notify.assert_called_once_with("hello world") + assert result["success"] is True + + @patch("apps.handlers.notifier.send_telegram_notification", return_value=False) + def test_notify_failure(self, mock_notify): + result = run("notify", ["fail"], {}) + assert result["success"] is False + + +class TestExceptionHandling: + @patch("apps.handlers.bot_operations.get_status", side_effect=RuntimeError("boom")) + def test_exception_caught_and_returned(self, mock_status): + result = run("status", [], {}) + assert result["success"] is False + assert "boom" in result["error"]