import unittest from unittest.mock import patch from app.services.ops_playbook_service import ( get_recent_ops_playbook_runs, list_ops_playbook_run_events, preview_ops_playbook, ) class OpsPlaybookServiceTests(unittest.TestCase): def test_onboarding_bootstrap_preview_uses_control_plane(self) -> None: ok, message, data = preview_ops_playbook( { "playbook_key": "onboarding.bootstrap", "node_codes": ["mainland-worker-02"], } ) self.assertTrue(ok, message) self.assertEqual("playbook 预览生成成功", message) self.assertEqual("onboarding.bootstrap", data["playbook"]["key"]) self.assertEqual("control-plane", data["execution_mode"]) self.assertEqual("控制面", data["execution_mode_label"]) self.assertEqual(["control-plane"], data["playbook"]["execution_modes"]) self.assertEqual("控制面", data["playbook"]["default_execution_mode_label"]) self.assertEqual("控制面", data["playbook"]["execution_mode_options"][0]["label"]) self.assertTrue(data["auto_approve"]) self.assertEqual(1, data["step_count"]) step = data["steps"][0] self.assertEqual("bootstrap", step["step_key"]) self.assertEqual("node.bootstrap", step["template_key"]) self.assertEqual("node.bootstrap", step["action"]) self.assertEqual("control-plane", step["execution_mode"]) self.assertEqual(1, step["expected_jobs"]) def test_scene_logs_key_preview_allows_ssh_execution_mode(self) -> None: ok, message, data = preview_ops_playbook( { "playbook_key": "scene.logs.key", "node_codes": ["mainland-worker-01"], "execution_mode": "ssh", } ) self.assertTrue(ok, message) self.assertEqual("playbook 预览生成成功", message) self.assertEqual("ssh", data["execution_mode"]) self.assertEqual("SSH", data["execution_mode_label"]) self.assertEqual(["remote-agent", "ssh"], data["playbook"]["execution_modes"]) self.assertEqual("ssh", data["steps"][0]["execution_mode"]) self.assertEqual("SSH", data["steps"][0]["execution_mode_label"]) def test_onboarding_bootstrap_preview_rejects_unsupported_execution_mode(self) -> None: ok, message, data = preview_ops_playbook( { "playbook_key": "onboarding.bootstrap", "node_codes": ["mainland-worker-02"], "execution_mode": "ssh", } ) self.assertFalse(ok) self.assertEqual("该 playbook 仅支持 control-plane", message) self.assertEqual(["control-plane"], data["playbook"]["execution_modes"]) @patch("app.services.ops_playbook_service.list_managed_nodes_with_agent_state") @patch("app.services.ops_playbook_service.list_ops_jobs") def test_recent_onboarding_bootstrap_run_focus_summary_reflects_agent_state( self, mock_list_ops_jobs, mock_list_managed_nodes_with_agent_state, ) -> None: mock_list_ops_jobs.return_value = [ { "id": 31, "job_code": "ops-bootstrap-31", "action": "node.bootstrap", "status": "success", "target_node_code": "mainland-worker-02", "created_at": "2026-04-17 10:00:00", "updated_at": "2026-04-17 10:01:00", "metadata": { "playbook": { "run_code": "pbr-bootstrap-1", "key": "onboarding.bootstrap", "title": "生成节点接入工单", "group_key": "onboarding", "group_title": "接管验收", "requested_by": "tester", "execution_mode": "control-plane", "step_key": "bootstrap", "step_title": "生成 Node Agent 接入工单", "step_order": 1, "step_count": 1, "expected_jobs_total": 1, "target_nodes_total": 1, "template_key": "node.bootstrap", } }, "result": { "bootstrap_plan": { "bootstrap_script_name": "bootstrap-node-agent-mainland-worker-02.sh", } }, } ] mock_list_managed_nodes_with_agent_state.return_value = { "nodes": [ { "node_code": "mainland-worker-02", "agent_state": "pending_bootstrap", "is_managed": True, } ] } payload = get_recent_ops_playbook_runs(limit=10, scan_limit=20) self.assertEqual(1, len(payload["runs"])) run = payload["runs"][0] self.assertEqual("success", run["status"]) self.assertEqual("成功", run["status_label"]) self.assertEqual("控制面", run["execution_mode_label"]) self.assertEqual("warning", run["focus_level"]) self.assertIn("待执行 bootstrap 脚本", run["focus_summary"]) self.assertEqual(run["focus_summary"], run["summary_text"]) self.assertEqual(1, run["steps_total"]) self.assertEqual("pbr-bootstrap-1", run["focus_ref"]["run_code"]) self.assertEqual("生成 Node Agent 接入工单", run["steps"][0]["title"]) self.assertEqual("成功", run["steps"][0]["status_label"]) self.assertTrue(run["steps"][0]["summary_text"]) @patch("app.services.ops_playbook_service.list_managed_nodes_with_agent_state") @patch("app.services.ops_playbook_service.list_ops_job_events_for_jobs") @patch("app.services.ops_playbook_service.list_ops_jobs") def test_playbook_run_events_add_contract_aliases( self, mock_list_ops_jobs, mock_list_ops_job_events_for_jobs, mock_list_managed_nodes_with_agent_state, ) -> None: mock_list_managed_nodes_with_agent_state.return_value = {"nodes": []} mock_list_ops_jobs.return_value = [ { "id": 41, "job_code": "ops-inspection-41", "action": "logs.collect", "status": "running", "target_node_code": "mainland-worker-01", "created_at": "2026-04-18 06:30:00", "updated_at": "2026-04-18 06:31:00", "metadata": { "playbook": { "run_code": "pbr-inspection-1", "key": "inspection.standard", "title": "标准巡检", "group_key": "diagnostics", "group_title": "标准巡检", "requested_by": "tester", "execution_mode": "remote-agent", "step_key": "worker_logs", "step_title": "收集 Worker 日志", "step_order": 2, "step_count": 3, "expected_jobs_total": 3, "target_nodes_total": 1, "template_key": "logs.collect", } }, } ] mock_list_ops_job_events_for_jobs.return_value = [ { "id": 501, "event_key": "job-event:501", "job_id": 41, "node_code": "mainland-worker-01", "event_type": "job_running", "level": "info", "level_label": "信息", "message": "开始收集 Worker 日志", "summary": "开始收集 Worker 日志", "summary_text": "远端 Agent 已开始收集 Worker 日志", "focus_ref": { "kind": "ops_job", "job_id": 41, "job_code": "ops-inspection-41", "action": "logs.collect", "target_node_code": "mainland-worker-01", "event_key": "job-event:501", "step_key": "worker_logs", }, "payload": {}, "created_at": "2026-04-18 06:31:00", "occurred_at": "2026-04-18 06:31:00", "ui_intent": {"kind": "job_detail", "job_id": 41, "focus_event_id": 501}, } ] payload = list_ops_playbook_run_events("pbr-inspection-1", limit=10) self.assertEqual("inspection.standard", payload["playbook_run"]["playbook_key"]) self.assertEqual(1, len(payload["events"])) event = payload["events"][0] self.assertEqual("收口中", event["job_status_label"]) self.assertEqual("远端 Agent 已开始收集 Worker 日志", event["summary_text"]) self.assertEqual("worker_logs", event["focus_ref"]["focus_step_key"]) self.assertEqual("job-event:501", event["focus_ref"]["event_key"]) self.assertEqual("ops_job", event["source_focus_ref"]["kind"]) self.assertEqual("job-event:501", event["source_focus_ref"]["event_key"]) summary = payload["summary"] self.assertEqual(1, summary["returned_total"]) self.assertEqual({"job_running": 1}, summary["event_type_counts"]) self.assertEqual({"running": 1}, summary["job_status_counts"]) self.assertEqual(10, summary["filters"]["limit"]) if __name__ == "__main__": unittest.main()