import unittest from unittest.mock import patch from app.services.ops_service import preview_driver_action class OpsServiceDriverActionPreviewTests(unittest.TestCase): def test_preview_driver_action_builds_safe_auto_contract_preview(self) -> None: ok, message, data = preview_driver_action( { "source_label": "overview-recommendation", "title": "先补现场日志回传", "summary": "优先打开关键日志回传。", "reason": "当前还看不到现场过程日志。", "executor_kind": "driver_action", "focus_ref": { "kind": "execution_scene", "scene_key": "scene.logs", }, "primary_focus_ref": { "kind": "execution_scene", "scene_key": "scene.logs", }, "primary_label": "开启关键回传", "primary_action_code": "enable_log_sync_key", "primary_action_payload": {}, "requested_by": "web-ui/ops-center", } ) self.assertTrue(ok, message) self.assertEqual("驾驶动作预览已生成", message) self.assertEqual("ops_driver_contract", data["contract_key"]) self.assertEqual("/api/v1/ops/driver-feed", data["contract_primary_endpoint"]) self.assertEqual("/api/v1/ops/driver-actions/preview", data["preview_endpoint"]) self.assertEqual("driver_action", data["executor_kind"]) self.assertEqual( "ops-driver -> driver-action execute -> playbook / ui-intent / settings", data["execution_chain"], ) primary = data["primary"] self.assertEqual("开启关键回传", primary["label"]) self.assertEqual("enable_log_sync_key", primary["action_code"]) self.assertEqual("/api/v1/ops/driver-actions/execute", primary["target_api"]) self.assertEqual("可自动执行", primary["recommendation_label"]) self.assertEqual("safe_auto", primary["automation_level_label"]) self.assertEqual("low", primary["risk_level_label"]) self.assertEqual("execution_scene", data["focus_ref"]["kind"]) self.assertEqual("execution_scene", data["primary_focus_ref"]["kind"]) self.assertEqual("execution_scene", primary["focus_ref"]["kind"]) self.assertEqual( { "action_code": "enable_log_sync_key", "node_codes": [], "action_payload": { "focus_ref": { "kind": "execution_scene", "scene_key": "scene.logs", } }, "requested_by": "web-ui/ops-center", }, primary["request_payload"], ) self.assertFalse(data["secondary"]) def test_preview_driver_action_marks_missing_required_fields_as_blocked(self) -> None: ok, message, data = preview_driver_action( { "source_label": "driver-feed", "title": "打开全量现场取证", "summary": "当前需要进入取证工作区。", "reason": "需要明确 playbook 上下文。", "executor_kind": "driver_action", "primary_label": "打开取证", "primary_action_code": "open_playbook_dialog", "primary_action_payload": {}, "requested_by": "web-ui/ops-center", } ) self.assertTrue(ok, message) primary = data["primary"] self.assertEqual("当前阻断", primary["recommendation_label"]) self.assertTrue(primary["execution_profile"]["blocked"]) self.assertEqual(["playbook_key"], primary["execution_profile"]["missing_fields"]) def test_preview_driver_action_marks_release_launchpad_review_as_safe_auto(self) -> None: ok, message, data = preview_driver_action( { "source_label": "overview-recommendation", "title": "查看 Worker 预案", "summary": "当前建议先复核 Worker launchpad 预案。", "reason": "需要先判断目标节点、门禁和阻断缺口。", "executor_kind": "driver_action", "focus_ref": { "kind": "release_hub", "release_id": 88, "section": "release_launchpad", "mode": "worker", }, "primary_focus_ref": { "kind": "release_hub", "release_id": 88, "section": "release_launchpad", "mode": "worker", }, "primary_label": "查看 Worker 预案", "primary_action_code": "review_smart_rollout_preview", "primary_action_payload": { "channel": "stable", }, "requested_by": "web-ui/ops-center", } ) self.assertTrue(ok, message) self.assertEqual("驾驶动作预览已生成", message) primary = data["primary"] self.assertEqual("review_smart_rollout_preview", primary["action_code"]) self.assertEqual("可自动执行", primary["recommendation_label"]) self.assertEqual("safe_auto", primary["automation_level_label"]) self.assertEqual("low", primary["risk_level_label"]) self.assertFalse(primary["execution_profile"]["ui_only"]) self.assertEqual("/api/v1/ops/driver-actions/execute", primary["target_api"]) def test_preview_driver_action_marks_handover_gap_as_safe_auto(self) -> None: ok, message, data = preview_driver_action( { "source_label": "overview-recommendation", "title": "先补接管缺口", "summary": "当前存在待接管节点。", "reason": "需要先查看首台缺口节点的接管详情。", "executor_kind": "driver_action", "focus_ref": { "kind": "managed_node", "node_code": "mainland-worker-01", }, "primary_focus_ref": { "kind": "managed_node", "node_code": "mainland-worker-01", }, "primary_label": "查看接管详情", "primary_action_code": "handover_first_gap", "primary_action_payload": { "node_code": "mainland-worker-01", }, "requested_by": "web-ui/ops-center", } ) self.assertTrue(ok, message) primary = data["primary"] self.assertEqual("handover_first_gap", primary["action_code"]) self.assertEqual("可自动执行", primary["recommendation_label"]) self.assertEqual("safe_auto", primary["automation_level_label"]) self.assertEqual("low", primary["risk_level_label"]) self.assertFalse(primary["execution_profile"]["ui_only"]) @patch("app.services.ops_service.get_ops_runbook") def test_preview_driver_action_supports_runbook_sequence_primary_and_secondary(self, mock_get_ops_runbook) -> None: mock_get_ops_runbook.return_value = { "control_sequences": [ { "key": "release_progression", "title": "版本发布与放量", "status": "ready", "status_label": "可发 Worker 灰度", "summary": "当前可以进入正式发布路径。", "reason": "统一回到 Release / Rollout 闭环。", "target_scope_label": "Release / Rollout / 2026.04.18", "step_titles": ["确认发布包", "发灰度"], "target_node_codes": ["mainland-controller-01"], "focus_ref": { "kind": "release_hub", "release_id": 7, "section": "release_launchpad", }, "primary_label": "Worker 灰度", "primary_action_code": "create_release_rollout_worker", "primary_action_payload": {"release_id": 7}, "primary_focus_ref": { "kind": "release_hub", "release_id": 7, "section": "release_launchpad", "mode": "worker", }, "secondary_label": "查看版本区", "secondary_action_code": "focus_release_hub", "secondary_action_payload": {"release_id": 7}, "secondary_focus_ref": { "kind": "release_hub", "release_id": 7, "section": "release_detail", }, } ] } ok, message, data = preview_driver_action( { "source_label": "driver-feed", "title": "版本发布与放量", "summary": "当前可以进入正式发布路径。", "reason": "统一回到 Release / Rollout 闭环。", "executor_kind": "runbook_sequence", "sequence_key": "release_progression", "primary_label": "Worker 灰度", "requested_by": "web-ui/ops-center", } ) self.assertTrue(ok, message) self.assertEqual("驾驶动作预览已生成", message) self.assertEqual("runbook_sequence", data["executor_kind"]) self.assertEqual("runbook sequence", data["executor_kind_label"]) self.assertEqual( "ops-driver -> runbook-sequence -> resolved driver action / ui-intent", data["execution_chain"], ) self.assertEqual("create_release_rollout_worker", data["resolved_primary"]["driver_action_code"]) self.assertEqual("release_launchpad", data["focus_ref"]["section"]) self.assertEqual("worker", data["primary_focus_ref"]["mode"]) self.assertEqual("release_detail", data["secondary_focus_ref"]["section"]) primary = data["primary"] self.assertEqual("/api/v1/ops/runbook/sequences/release_progression/execute", primary["target_api"]) self.assertEqual("建议确认后执行", primary["recommendation_label"]) self.assertEqual("guarded_auto", primary["automation_level_label"]) self.assertEqual("worker", primary["focus_ref"]["mode"]) self.assertEqual( { "secondary": False, "requested_by": "web-ui/ops-center", "node_codes": ["mainland-controller-01"], "action_payload": { "release_id": 7, "focus_ref": { "kind": "release_hub", "release_id": 7, "section": "release_launchpad", "mode": "worker", }, }, }, primary["request_payload"], ) secondary = data["secondary"] self.assertEqual("查看版本区", secondary["label"]) self.assertEqual("focus_release_hub", secondary["action_code"]) self.assertEqual("release_detail", secondary["focus_ref"]["section"]) self.assertEqual( { "secondary": True, "requested_by": "web-ui/ops-center", "node_codes": ["mainland-controller-01"], "action_payload": { "release_id": 7, "focus_ref": { "kind": "release_hub", "release_id": 7, "section": "release_detail", }, }, }, secondary["request_payload"], ) self.assertEqual("可自动执行", secondary["recommendation_label"]) if __name__ == "__main__": unittest.main()