import unittest from unittest.mock import patch from app.services.ops_service import get_ops_driver_feed class OpsServiceDriverFeedTests(unittest.TestCase): @patch("app.services.ops_service.get_ops_go_live_summary") @patch("app.services.ops_service.get_ops_runbook") @patch("app.services.ops_service.get_ops_overview") def test_driver_feed_merges_priority_recommendation_driver_cards_and_runbook_sequences( self, mock_get_ops_overview, mock_get_ops_runbook, mock_get_ops_go_live_summary, ) -> None: mock_get_ops_go_live_summary.return_value = { "go_live_status": "blocked", "publish_ready": False, "publish_status": "blocked", "publish_status_label": "暂不可发布", "publish_summary": "当前仍有硬阻断项，不能直接进入正式发版。", "operator_title": "先补接管缺口", "next_step_action_code": "handover_first_gap", "launchpad_recommended_target_node_code": "mainland-controller-01", "launchpad_recommended_recovery_label": "签发接入工单", "launchpad_recommended_recovery_summary": "当前节点仍处于接入阶段，建议先签发 onboarding.bootstrap。", "launchpad_onboarding_bootstrap_pending_nodes": 1, "launchpad_onboarding_acceptance_ready_nodes": 1, "blocking_reasons": ["remote_agent_ready=0"], "warnings": ["log_sync_waiting_sample=0/2"], } mock_get_ops_overview.return_value = { "execution_scene": { "participation_summary": { "participating_nodes": 2, "dispatch_active_node_codes": ["mainland-worker-01", "mainland-controller-01"], }, "participating_nodes": [ {"node_code": "mainland-worker-01"}, {"node_code": "mainland-controller-01"}, ], "log_sync": { "enabled": False, "mode": "key", "line_count": 0, "source_node_count": 0, "participating_node_count": 2, "covered_participating_node_count": 0, "missing_participating_nodes": ["mainland-worker-01", "mainland-controller-01"], }, }, "recommendation": { "key": "fix-handover-gap", "priority": "先补接管缺口", "title": "先补接管缺口", "summary": "当前应该先把接管链路补齐。", "reason": "否则后续批量动作无法稳定下发。", "level_label": "最高优先", "tag_type": "danger", "primary_label": "纳管首台节点", "secondary_label": "查看目标节点", "primary_action_code": "handover_first_gap", "secondary_action_code": "view_first_gap", "node_codes": ["mainland-worker-02"], "primary_node_codes": ["mainland-worker-02"], "secondary_node_codes": ["mainland-worker-02"], "primary_action_payload": {"requested_by": "test"}, "secondary_action_payload": {}, "reasons": ["当前存在接管缺口。"], }, "activity_stream": { "items": [ { "kind": "ops_job", "activity_key": "ops-job:501", "title": "logs.collect", "subtitle": "ops-logs-501", "summary_text": "远端 Agent 已回传 120 行 Worker 日志，可直接继续分析。", "summary": "远端 Agent 已回传 120 行 Worker 日志，可直接继续分析。", "status": "running", "status_label": "执行中", "meta_text": "ops-logs-501 / 执行 远端 Agent", "occurred_at": "2026-04-18 09:18:00", "target_node_codes": ["mainland-worker-01"], "focus_ref": { "kind": "ops_job", "job_id": 501, "job_code": "ops-logs-501", "action": "logs.collect", "target_node_code": "mainland-worker-01", }, "source_focus_ref": { "kind": "ops_job_event", "job_id": 501, "event_key": "job-event:501", }, "ui_intent": { "kind": "job_events", "job_id": 501, "job_code": "ops-logs-501", "action": "logs.collect", "target_node_code": "mainland-worker-01", }, } ] }, "driver_recommendations": [ { "key": "fix-handover-gap", "title": "先补接管缺口", "summary": "重复建议，不应再次进入 feed。", "reason": "duplicate", "level_label": "最高优先", "tag_type": "danger", "primary_label": "纳管首台节点", "primary_action_code": "handover_first_gap", }, { "key": "inspect-scene-log-gap", "title": "排查现场日志样本缺口", "summary": "现场日志样本还没有形成。", "reason": "先看 Worker 日志。", "meta_text": "参与节点 2 台 / 已回传来源 0 台", "level_label": "推荐", "tag_type": "warning", "primary_label": "看 Worker 日志", "secondary_label": "执行标准巡检", "primary_action_code": "open_worker_logs", "secondary_action_code": "run_standard_inspection", "node_codes": ["mainland-worker-01", "mainland-controller-01"], "primary_node_codes": ["mainland-worker-01", "mainland-controller-01"], "secondary_node_codes": ["mainland-worker-01", "mainland-controller-01"], "primary_action_payload": {"lines": 120}, "secondary_action_payload": {}, }, ], } mock_get_ops_runbook.return_value = { "control_sequences": [ { "key": "release_progression", "title": "版本发布与放量", "summary": "当前可进入发布路径。", "reason": "正式运维时应统一落回 Release / Rollout。", "status": "warning", "status_label": "待处理", "tag_type": "warning", "target_scope_label": "Release / Rollout", "target_node_codes": ["mainland-controller-01"], "focus_ref": { "kind": "release_hub", "release_id": 7, "section": "release_launchpad", }, "primary_label": "补执行面接管", "secondary_label": "查看版本区", "primary_action_code": "fix_managed_nodes", "secondary_action_code": "focus_release_hub", "primary_focus_ref": { "kind": "release_hub", "release_id": 7, "section": "default_rollout_gate", }, "secondary_focus_ref": { "kind": "release_hub", "release_id": 7, "section": "release_detail", }, "step_titles": ["同步集群节点", "纳管首台缺口", "补齐 Agent 接入"], "primary_resolution": { "ok": True, "driver_action_code": "fix_managed_nodes", "driver_action_label": "补执行面接管", "driver_node_codes": ["mainland-controller-01"], "driver_action_payload": {"channel": "stable"}, "resolved_at": "2026-04-18T03:00:00", }, "secondary_resolution": { "ok": True, "driver_action_code": "focus_release_hub", "driver_action_label": "查看版本区", "driver_node_codes": [], "driver_action_payload": {"release_id": 7}, }, } ] } payload = get_ops_driver_feed() self.assertEqual(3, payload["summary"]["total"]) self.assertEqual("blocked", payload["go_live_summary"]["go_live_status"]) self.assertEqual("blocked", payload["summary"]["go_live_status"]) self.assertEqual("先补接管缺口", payload["summary"]["operator_title"]) self.assertEqual("blocked", payload["summary"]["publish_status"]) self.assertFalse(payload["summary"]["publish_ready"]) self.assertEqual("暂不可发布", payload["summary"]["publish_status_label"]) self.assertIn("发布闸门：暂不可发布", payload["headline"]) self.assertIn("接入缺口：签发接入工单：mainland-controller-01", payload["headline"]) self.assertEqual("handover_first_gap", payload["go_live_summary"]["next_step_action_code"]) self.assertEqual("mainland-controller-01", payload["summary"]["launchpad_recommended_target_node_code"]) self.assertEqual("签发接入工单", payload["summary"]["launchpad_recommended_recovery_label"]) self.assertEqual( "当前节点仍处于接入阶段，建议先签发 onboarding.bootstrap。", payload["summary"]["launchpad_recommended_recovery_summary"], ) self.assertEqual(1, payload["summary"]["launchpad_onboarding_bootstrap_pending_nodes"]) self.assertEqual(1, payload["summary"]["launchpad_onboarding_acceptance_ready_nodes"]) self.assertEqual(2, payload["summary"]["do_now_total"]) self.assertEqual(1, payload["summary"]["standard_path_total"]) self.assertEqual("ops_driver_contract", payload["contract_navigation"]["primary_contract_key"]) self.assertIn("ops_playbook_contract", payload["contract_navigation"]["contract_keys"]) self.assertIn("ops_observability_contract", payload["contract_navigation"]["contract_keys"]) self.assertEqual("priority:fix-handover-gap", payload["top_recommendation"]["key"]) self.assertEqual("ops_driver_contract", payload["top_recommendation"]["contract_navigation"]["primary_contract_key"]) self.assertEqual(2, len(payload["driver_recommendations"])) self.assertEqual(1, len(payload["runbook_sequences"])) self.assertEqual(1, len(payload["activity_focus"])) self.assertEqual( ["priority_recommendation", "driver_recommendation", "runbook_sequence"], [item["kind"] for item in payload["entries"]], ) self.assertEqual(1, payload["summary"]["activity_focus_total"]) self.assertEqual("disabled", payload["summary"]["scene_log_status"]) self.assertEqual("blocked", payload["summary"]["launch_status"]) self.assertFalse(payload["summary"]["launch_ready"]) self.assertEqual(2, payload["summary"]["preview_only_total"]) self.assertEqual(3, payload["summary"]["backend_handled_total"]) self.assertEqual("blocked", payload["automation_coverage"]["launch_status"]) self.assertEqual(2, payload["automation_coverage"]["safe_auto_total"]) self.assertEqual(0, payload["automation_coverage"]["guarded_auto_total"]) self.assertEqual(2, payload["automation_coverage"]["auto_execute_total"]) self.assertEqual(1, payload["automation_coverage"]["resolve_first_total"]) self.assertEqual(2, payload["automation_coverage"]["preview_only_total"]) self.assertEqual(3, payload["automation_coverage"]["backend_handled_total"]) self.assertEqual(1, payload["automation_coverage"]["human_dependency_total"]) self.assertFalse(payload["automation_coverage"]["launch_ready"]) priority_entry = payload["entries"][0] self.assertEqual("do_now", priority_entry["lane"]) self.assertEqual("driver_action", priority_entry["executor_kind"]) self.assertEqual("handover_first_gap", priority_entry["primary_action_code"]) self.assertEqual("ops_driver_contract", priority_entry["contract_navigation"]["primary_contract_key"]) activity_focus = payload["activity_focus"][0] self.assertEqual("ops_job", activity_focus["activity_kind"]) self.assertEqual("远端 Agent 已回传 120 行 Worker 日志，可直接继续分析。", activity_focus["summary"]) self.assertEqual("focus_latest_job_events", activity_focus["focus_action_code"]) self.assertEqual("job_events", activity_focus["ui_intent_kind"]) self.assertEqual("ops_job_event", activity_focus["source_focus_ref"]["kind"]) self.assertTrue(activity_focus["observation_only"]) self.assertEqual("ops_observability_contract", activity_focus["contract_navigation"]["primary_contract_key"]) self.assertIn("ops_stack_diagnosis_contract", activity_focus["contract_navigation"]["contract_keys"]) scene_log_observation = payload["scene_log_observation"] self.assertEqual("disabled", scene_log_observation["status"]) self.assertEqual("mainland-worker-01", scene_log_observation["target_node_code"]) self.assertEqual("node_scene_log", scene_log_observation["focus_ref"]["kind"]) self.assertEqual( "ops_observability_contract", scene_log_observation["contract_navigation"]["primary_contract_key"], ) runbook_entry = payload["entries"][2] self.assertEqual("runbook_sequence", runbook_entry["kind"]) self.assertEqual("standard_path", runbook_entry["lane"]) self.assertEqual("runbook_sequence", runbook_entry["executor_kind"]) self.assertEqual("release_progression", runbook_entry["sequence_key"]) self.assertEqual("fix_managed_nodes", runbook_entry["primary_action_code"]) self.assertEqual("focus_release_hub", runbook_entry["secondary_action_code"]) self.assertIn("当前解析：补执行面接管", runbook_entry["detail_lines"][0]) self.assertEqual("release_hub", runbook_entry["focus_ref"]["kind"]) self.assertEqual("release_launchpad", runbook_entry["focus_ref"]["section"]) self.assertEqual("default_rollout_gate", runbook_entry["primary_focus_ref"]["section"]) self.assertEqual("release_detail", runbook_entry["secondary_focus_ref"]["section"]) self.assertEqual("default_rollout_gate", runbook_entry["primary_action_payload"]["focus_ref"]["section"]) self.assertEqual("release_detail", runbook_entry["secondary_action_payload"]["focus_ref"]["section"]) self.assertEqual("ops_driver_contract", runbook_entry["contract_navigation"]["primary_contract_key"]) self.assertIn("ops_playbook_contract", runbook_entry["contract_navigation"]["contract_keys"]) self.assertIn("release_hub_contract", runbook_entry["contract_navigation"]["contract_keys"]) @patch("app.services.ops_service.get_ops_go_live_summary") @patch("app.services.ops_service.get_ops_runbook") @patch("app.services.ops_service.get_ops_overview") def test_driver_feed_preserves_bootstrap_run_as_top_recommendation( self, mock_get_ops_overview, mock_get_ops_runbook, mock_get_ops_go_live_summary, ) -> None: mock_get_ops_go_live_summary.return_value = { "go_live_status": "attention", "publish_ready": False, "publish_status": "attention", "publish_status_label": "可发布但建议先复核", "publish_summary": "当前没有硬阻断，但仍有上线前关注项，建议先完成复核再正式发版。", "operator_title": "先跑接入收口", "next_step_action_code": "bootstrap_run", "launchpad_recommended_target_node_code": "mainland-worker-01", "launchpad_recommended_recovery_label": "跑接入收口", "launchpad_recommended_recovery_summary": "mainland-worker-01 还缺接入收口，先执行 bootstrap。", "launchpad_onboarding_bootstrap_pending_nodes": 1, "launchpad_onboarding_acceptance_ready_nodes": 0, "blocking_reasons": [], "warnings": ["remote_agent_ready=1/2"], } mock_get_ops_overview.return_value = { "recommendation": { "key": "handover-gap-bootstrap_run", "priority": "先补接管缺口", "title": "先补接管缺口", "summary": "Launchpad 已明确收敛到首台节点接入收口。", "reason": "mainland-worker-01 还缺接入收口，先执行 bootstrap。", "level_label": "最高优先", "tag_type": "warning", "primary_label": "跑接入收口", "secondary_label": "查看目标节点", "primary_action_code": "bootstrap_run", "secondary_action_code": "view_first_gap", "node_codes": ["mainland-worker-01"], "primary_node_codes": ["mainland-worker-01"], "secondary_node_codes": ["mainland-worker-01"], "primary_action_payload": {"node_code": "mainland-worker-01"}, "secondary_action_payload": {}, "focus_ref": {"kind": "managed_node", "node_code": "mainland-worker-01"}, "primary_focus_ref": {"kind": "managed_node", "node_code": "mainland-worker-01"}, "secondary_focus_ref": {"kind": "managed_node", "node_code": "mainland-worker-01"}, }, "driver_recommendations": [], "activity_stream": {"items": []}, "execution_scene": {}, } mock_get_ops_runbook.return_value = {"control_sequences": []} payload = get_ops_driver_feed() self.assertEqual("priority:handover-gap-bootstrap_run", payload["top_recommendation"]["key"]) self.assertEqual("attention", payload["summary"]["publish_status"]) self.assertEqual("可发布但建议先复核", payload["summary"]["publish_status_label"]) self.assertEqual("mainland-worker-01", payload["summary"]["launchpad_recommended_target_node_code"]) self.assertEqual("跑接入收口", payload["summary"]["launchpad_recommended_recovery_label"]) self.assertEqual(1, payload["summary"]["launchpad_onboarding_bootstrap_pending_nodes"]) self.assertEqual(0, payload["summary"]["launchpad_onboarding_acceptance_ready_nodes"]) self.assertEqual("attention", payload["summary"]["launch_status"]) self.assertFalse(payload["summary"]["launch_ready"]) self.assertEqual("bootstrap_run", payload["top_recommendation"]["primary_action_code"]) self.assertEqual("mainland-worker-01", payload["top_recommendation"]["primary_node_codes"][0]) self.assertIn("ops_agent_protocol", payload["top_recommendation"]["contract_navigation"]["contract_keys"]) self.assertEqual("ops_driver_contract", payload["top_recommendation"]["contract_navigation"]["primary_contract_key"]) self.assertEqual(0, payload["automation_coverage"]["safe_auto_total"]) self.assertEqual(1, payload["automation_coverage"]["mixed_total"]) self.assertEqual(0, payload["automation_coverage"]["preview_only_total"]) self.assertEqual(1, payload["automation_coverage"]["backend_handled_total"]) if __name__ == "__main__": unittest.main()