import unittest from unittest.mock import patch from app.services.ops_service import get_ops_runbook class OpsServiceRunbookTests(unittest.TestCase): @patch("app.services.ops_service.get_release_launchpad") @patch("app.services.ops_service.get_latest_release") @patch("app.services.ops_service.list_managed_nodes_with_agent_state") @patch("app.services.ops_service.get_runtime_status") def test_ops_runbook_includes_control_sequences( self, mock_get_runtime_status, mock_list_managed_nodes_with_agent_state, mock_get_latest_release, mock_get_release_launchpad, ) -> None: mock_get_runtime_status.return_value = { "readiness": { "status": "ready", "summary": "当前多机现场已进入标准作业状态", }, "worker": { "running": True, }, "sync_agent": { "running": True, }, "detect": { "participation_summary": { "summary": "worker 正在执行", } }, } mock_list_managed_nodes_with_agent_state.return_value = { "nodes": [ { "node_code": "mainland-controller-01", "role": "control", "is_managed": True, "is_enabled": True, "is_agent_online": True, "cluster_status": "online", "cluster_is_effective_worker": True, "is_current_participant": False, }, { "node_code": "mainland-worker-01", "role": "worker", "is_managed": True, "is_enabled": True, "is_agent_online": True, "cluster_status": "busy", "cluster_is_effective_worker": True, "is_current_participant": True, }, ] } mock_get_latest_release.side_effect = lambda channel="stable": ( { "id": 15, "release_version": "2026.04.17", "channel": "stable", } if channel == "stable" else {} ) mock_get_release_launchpad.return_value = { "channel": "stable", "latest_release": { "id": 15, "release_version": "2026.04.17", "channel": "stable", }, "launchpad_status": { "status": "ready", "status_label": "可发 Worker 灰度", "summary": "最新发布包、Release 记录与 Worker 智能 Rollout 预检均已收口。", "recommended_action_code": "publish_latest_worker", "focus_ref": { "kind": "release_hub", "release_id": 15, "section": "release_launchpad", }, }, "worker_rollout_preview": { "message": "智能 Rollout 预检通过", "smart_rollout": { "available": True, "summary": "已按 Worker 模式预选 1 台节点。", "target_node_codes": ["mainland-worker-01"], }, }, "control_rollout_preview": { "message": "", "smart_rollout": { "available": False, "target_node_codes": [], }, }, } runbook = get_ops_runbook() mock_list_managed_nodes_with_agent_state.assert_called_once_with( participation_payload=mock_get_runtime_status.return_value["detect"] ) self.assertIn("GET /api/v1/ops/releases/launchpad", runbook["fast_checks"]) self.assertEqual("ready", runbook["release_launchpad"]["launchpad_status"]["status"]) sequences = {item["key"]: item for item in runbook["control_sequences"]} self.assertIn("node_onboarding", sequences) self.assertIn("scene_observe", sequences) self.assertIn("standard_inspection", sequences) self.assertIn("release_progression", sequences) self.assertEqual("bootstrap_run", sequences["node_onboarding"]["primary_action_code"]) self.assertEqual("run_acceptance", sequences["node_onboarding"]["secondary_action_code"]) self.assertEqual("run_standard_inspection", sequences["standard_inspection"]["primary_action_code"]) self.assertEqual("open_diagnostics", sequences["standard_inspection"]["secondary_action_code"]) self.assertEqual(["mainland-worker-01"], sequences["scene_observe"]["target_node_codes"]) self.assertEqual("publish_latest_worker", sequences["release_progression"]["primary_action_code"]) self.assertEqual(15, sequences["release_progression"]["primary_action_payload"]["release_id"]) self.assertEqual(["mainland-worker-01"], sequences["release_progression"]["target_node_codes"]) self.assertTrue(sequences["release_progression"]["primary_resolution"]["ok"]) self.assertEqual("publish_latest_worker", sequences["release_progression"]["primary_resolution"]["driver_action_code"]) self.assertEqual(["mainland-worker-01"], sequences["release_progression"]["primary_resolution"]["driver_node_codes"]) self.assertEqual("release_hub", sequences["release_progression"]["focus_ref"]["kind"]) self.assertEqual("release_launchpad", sequences["release_progression"]["focus_ref"]["section"]) self.assertEqual("worker", sequences["release_progression"]["primary_focus_ref"]["mode"]) self.assertEqual("worker", sequences["release_progression"]["primary_action_payload"]["focus_ref"]["mode"]) self.assertEqual("release_detail", sequences["release_progression"]["secondary_focus_ref"]["section"]) self.assertTrue(sequences["scene_observe"]["secondary_resolution"]["ok"]) self.assertEqual("run_scene_logs_full", sequences["scene_observe"]["secondary_resolution"]["driver_action_code"]) @patch("app.services.ops_service.get_release_launchpad") @patch("app.services.ops_service.get_latest_release") @patch("app.services.ops_service.list_managed_nodes_with_agent_state") @patch("app.services.ops_service.get_runtime_status") def test_ops_runbook_release_progression_can_fall_back_to_bootstrap_run( self, mock_get_runtime_status, mock_list_managed_nodes_with_agent_state, mock_get_latest_release, mock_get_release_launchpad, ) -> None: mock_get_runtime_status.return_value = { "readiness": { "status": "attention", "summary": "当前还需补执行面接管", }, "worker": {"running": True}, "sync_agent": {"running": True}, "detect": {"participation_summary": {"summary": ""}}, } mock_list_managed_nodes_with_agent_state.return_value = { "nodes": [ { "node_code": "mainland-worker-02", "role": "worker", "is_managed": False, "is_enabled": False, "is_agent_online": False, "cluster_status": "online", "cluster_is_effective_worker": True, "is_current_participant": False, } ] } mock_get_latest_release.return_value = {} mock_get_release_launchpad.return_value = { "channel": "stable", "latest_release": {}, "launchpad_status": { "status": "attention", "status_label": "待补执行面", "summary": "当前节点仍处于接入阶段，建议先签发 onboarding.bootstrap。", "recommended_action_code": "bootstrap_run", "focus_ref": { "kind": "release_hub", "release_id": 19, "section": "release_launchpad", }, }, "worker_rollout_preview": { "message": "当前没有满足角色条件的在线节点可用于智能 Rollout。", "smart_rollout": { "available": False, "target_node_codes": [], }, "policy_preview": { "release_gate": { "operational_readiness": { "rows": [ { "node_code": "mainland-worker-02", "is_managed": False, "is_enabled": False, "is_agent_online": False, "remote_agent_ready": False, "inspection_status": "missing", "recovery_action": "bootstrap_run", "recovery_summary": "当前节点仍处于接入阶段，建议先签发 onboarding.bootstrap。", "agent_reason": "节点尚未纳管，当前不能稳定下发 remote-agent 动作。", } ] } } }, }, "control_rollout_preview": { "message": "", "smart_rollout": { "available": False, "target_node_codes": [], }, }, } runbook = get_ops_runbook() sequences = {item["key"]: item for item in runbook["control_sequences"]} self.assertEqual("bootstrap_run", sequences["release_progression"]["primary_action_code"]) self.assertEqual(["mainland-worker-02"], sequences["release_progression"]["target_node_codes"]) self.assertEqual("待补执行面", sequences["release_progression"]["status_label"]) self.assertTrue(sequences["release_progression"]["primary_resolution"]["ok"]) self.assertEqual("bootstrap_run", sequences["release_progression"]["primary_resolution"]["driver_action_code"]) self.assertEqual("default_rollout_gate", sequences["release_progression"]["primary_focus_ref"]["section"]) self.assertEqual("default_rollout_gate", sequences["release_progression"]["primary_action_payload"]["focus_ref"]["section"]) @patch("app.services.ops_service.get_release_launchpad") @patch("app.services.ops_service.get_latest_release") @patch("app.services.ops_service.list_managed_nodes_with_agent_state") @patch("app.services.ops_service.get_runtime_status") def test_ops_runbook_release_progression_falls_back_to_direct_worker_rollout_when_release_exists( self, mock_get_runtime_status, mock_list_managed_nodes_with_agent_state, mock_get_latest_release, mock_get_release_launchpad, ) -> None: mock_get_runtime_status.return_value = { "readiness": { "status": "ready", "summary": "当前多机现场已进入标准作业状态", }, "worker": {"running": True}, "sync_agent": {"running": True}, "detect": {"participation_summary": {"summary": ""}}, } mock_list_managed_nodes_with_agent_state.return_value = { "nodes": [ { "node_code": "mainland-worker-01", "role": "worker", "is_managed": True, "is_enabled": True, "is_agent_online": True, "cluster_status": "online", "cluster_is_effective_worker": True, "is_current_participant": False, } ] } mock_get_latest_release.return_value = { "id": 18, "release_version": "2026.04.18", "channel": "stable", } mock_get_release_launchpad.return_value = {} runbook = get_ops_runbook() sequences = {item["key"]: item for item in runbook["control_sequences"]} self.assertEqual("create_release_rollout_worker", sequences["release_progression"]["primary_action_code"]) self.assertEqual("focus_release_hub", sequences["release_progression"]["secondary_action_code"]) self.assertEqual(18, sequences["release_progression"]["primary_action_payload"]["release_id"]) self.assertEqual(["mainland-worker-01"], sequences["release_progression"]["target_node_codes"]) self.assertTrue(sequences["release_progression"]["primary_resolution"]["ok"]) self.assertEqual("create_release_rollout_worker", sequences["release_progression"]["primary_resolution"]["driver_action_code"]) if __name__ == "__main__": unittest.main()