from __future__ import annotations import unittest from unittest.mock import patch from app.services.ops_service import execute_codex_action, resolve_codex_action def _build_codex_brief_entry( *, key: str, executor_kind: str, primary_action_code: str, recommendation: str, automation_level: str, target_api: str, target_method: str = "POST", sequence_key: str = "", primary_action_payload: dict | None = None, secondary_action_code: str = "", secondary_action_payload: dict | None = None, focus_ref: dict | None = None, primary_focus_ref: dict | None = None, secondary_focus_ref: dict | None = None, ) -> dict: primary_action_payload = dict(primary_action_payload or {}) secondary_action_payload = dict(secondary_action_payload or {}) normalized_focus_ref = dict(focus_ref or {}) normalized_primary_focus_ref = dict(primary_focus_ref or {}) normalized_secondary_focus_ref = dict(secondary_focus_ref or {}) if normalized_primary_focus_ref and not dict(primary_action_payload.get("focus_ref") or {}): primary_action_payload["focus_ref"] = dict(normalized_primary_focus_ref) if normalized_secondary_focus_ref and not dict(secondary_action_payload.get("focus_ref") or {}): secondary_action_payload["focus_ref"] = dict(normalized_secondary_focus_ref) request_payload_preview = ( { "sequence_key": sequence_key, "secondary": False, "action_payload": primary_action_payload, } if executor_kind == "runbook_sequence" else { "action_code": primary_action_code, "node_codes": ["mainland-controller-01"], "action_payload": primary_action_payload, } ) return { "key": key, "title": "测试动作", "summary": "测试摘要", "reason": "测试原因", "meta_text": "meta", "executor_kind": executor_kind, "sequence_key": sequence_key, "focus_ref": normalized_focus_ref, "primary_focus_ref": normalized_primary_focus_ref, "secondary_focus_ref": normalized_secondary_focus_ref, "primary_label": "主动作", "secondary_label": "次动作" if secondary_action_code else "", "primary_action_code": primary_action_code, "secondary_action_code": secondary_action_code, "autopilot": { "status_text": "测试状态", "automation_level": automation_level, "recommendation": recommendation, "executor_mode_hint": "test-mode", "confirm_required": recommendation == "confirm_then_execute", "ui_only": recommendation == "open_ui", "blocked": recommendation == "blocked", "backend_handled": True, "risk_level": "low" if recommendation == "auto_execute" else "high", "reason": "autopilot reason", "target_method": target_method, "target_api": target_api, }, "primary_execution": { "label": "主动作", "node_codes": ["mainland-controller-01"], "action_payload": primary_action_payload, "request_payload_preview": request_payload_preview, }, "secondary_execution": ( { "label": "次动作", "node_codes": [], "action_payload": secondary_action_payload, } if secondary_action_code else {} ), } def _build_preview_data( *, executor_kind: str, action_code: str, recommendation: str, automation_level: str, target_api: str, request_payload: dict, sequence_key: str = "", focus_ref: dict | None = None, primary_focus_ref: dict | None = None, secondary_focus_ref: dict | None = None, ) -> dict: return { "executor_kind": executor_kind, "sequence_key": sequence_key, "execution_chain": "ops-driver -> preview", "focus_ref": dict(focus_ref or {}), "primary_focus_ref": dict(primary_focus_ref or {}), "secondary_focus_ref": dict(secondary_focus_ref or {}), "primary": { "label": "主动作", "action_code": action_code, "target_api": target_api, "target_method": "POST", "request_payload": dict(request_payload or {}), "focus_ref": dict(primary_focus_ref or {}), "execution_profile": { "action_code": action_code, "backend_handled": True, "automation_level": automation_level, "recommendation": recommendation, "executor_mode_hint": "test-mode", "risk_level": "low" if recommendation == "auto_execute" else "high", "confirm_required": recommendation == "confirm_then_execute", "ui_only": recommendation == "open_ui", "blocked": recommendation == "blocked", "requires_fields": [], "missing_fields": [], "reason": "preview reason", }, }, "secondary": {}, } def _build_codex_activity_focus_entry( *, key: str, activity_key: str, activity_kind: str, ui_intent_kind: str, focus_ref: dict | None = None, source_focus_ref: dict | None = None, target_node_codes: list[str] | None = None, ) -> dict: normalized_focus_ref = dict(focus_ref or {}) return { "key": key, "activity_key": activity_key, "activity_kind": activity_kind, "title": "观察焦点", "subtitle": "日志回传待确认", "summary": "该焦点属于观察项，应进入对应工作区查看上下文。", "status": "attention", "status_label": "待关注", "meta_text": "activity-meta", "occurred_at": "2026-04-18T06:00:00", "target_node_codes": list(target_node_codes or []), "focus_ref": normalized_focus_ref, "source_focus_ref": dict(source_focus_ref or {}), "ui_intent": { "kind": ui_intent_kind, "activity_key": activity_key, }, "ui_intent_kind": ui_intent_kind, "focus_action_code": "focus_activity_item", "focus_action_payload": { "activity_key": activity_key, "kind": activity_kind, "status": "attention", "ui_intent": { "kind": ui_intent_kind, "activity_key": activity_key, }, "focus_ref": normalized_focus_ref, }, "observation_only": True, } class OpsServiceCodexActionTests(unittest.TestCase): @patch("app.services.ops_service.preview_driver_action") @patch("app.services.ops_service.get_ops_codex_brief") def test_resolve_codex_action_returns_gate_for_safe_auto(self, mock_get_ops_codex_brief, mock_preview_driver_action) -> None: mock_get_ops_codex_brief.return_value = { "focus": {"entry_key": "recommendation:enable-log-sync"}, "summary": {"total": 1}, "entries": [ _build_codex_brief_entry( key="recommendation:enable-log-sync", executor_kind="driver_action", primary_action_code="enable_log_sync_key", recommendation="auto_execute", automation_level="safe_auto", target_api="/api/v1/ops/driver-actions/execute", focus_ref={ "kind": "execution_scene", "scene_key": "scene.logs", }, primary_focus_ref={ "kind": "execution_scene", "scene_key": "scene.logs", }, ) ], } mock_preview_driver_action.return_value = ( True, "ok", _build_preview_data( executor_kind="driver_action", action_code="enable_log_sync_key", recommendation="auto_execute", automation_level="safe_auto", target_api="/api/v1/ops/driver-actions/execute", request_payload={ "action_code": "enable_log_sync_key", "node_codes": ["mainland-controller-01"], "action_payload": { "focus_ref": { "kind": "execution_scene", "scene_key": "scene.logs", } }, "requested_by": "test/codex-action", }, focus_ref={ "kind": "execution_scene", "scene_key": "scene.logs", }, primary_focus_ref={ "kind": "execution_scene", "scene_key": "scene.logs", }, ), ) ok, message, data = resolve_codex_action({"requested_by": "test/codex-action"}) self.assertTrue(ok) self.assertEqual("Codex 驾驶动作解析完成", message) self.assertEqual("recommendation:enable-log-sync", data["entry_key"]) self.assertEqual("execute_now", data["gate"]["decision"]) self.assertTrue(data["gate"]["will_execute"]) self.assertEqual("/api/v1/ops/driver-actions/execute", data["gate"]["target_api"]) self.assertEqual("enable_log_sync_key", data["preview_request"]["primary_action_code"]) self.assertEqual("execution_scene", data["preview_request"]["focus_ref"]["kind"]) self.assertEqual("scene.logs", data["gate"]["focus_ref"]["scene_key"]) self.assertEqual("scene.logs", data["focus_ref"]["scene_key"]) @patch("app.services.ops_service.preview_driver_action") @patch("app.services.ops_service.get_ops_codex_brief") def test_resolve_codex_action_requires_confirmation_for_guarded_runbook( self, mock_get_ops_codex_brief, mock_preview_driver_action, ) -> None: mock_get_ops_codex_brief.return_value = { "focus": {"entry_key": "runbook:release_progression"}, "summary": {"total": 1}, "entries": [ _build_codex_brief_entry( key="runbook:release_progression", executor_kind="runbook_sequence", sequence_key="release_progression", primary_action_code="create_release_rollout_worker", recommendation="confirm_then_execute", automation_level="guarded_auto", target_api="/api/v1/ops/runbook/sequences/release_progression/execute", primary_action_payload={"release_id": 7}, secondary_action_code="focus_release_hub", secondary_action_payload={"release_id": 7}, focus_ref={ "kind": "release_hub", "release_id": 7, "section": "release_launchpad", }, primary_focus_ref={ "kind": "release_hub", "release_id": 7, "section": "release_launchpad", "mode": "worker", }, secondary_focus_ref={ "kind": "release_hub", "release_id": 7, "section": "release_detail", }, ) ], } mock_preview_driver_action.return_value = ( True, "ok", _build_preview_data( executor_kind="runbook_sequence", action_code="create_release_rollout_worker", recommendation="confirm_then_execute", automation_level="guarded_auto", target_api="/api/v1/ops/runbook/sequences/release_progression/execute", request_payload={ "secondary": False, "requested_by": "test/codex-action", "action_payload": { "release_id": 7, "focus_ref": { "kind": "release_hub", "release_id": 7, "section": "release_launchpad", "mode": "worker", }, }, }, sequence_key="release_progression", focus_ref={ "kind": "release_hub", "release_id": 7, "section": "release_launchpad", }, primary_focus_ref={ "kind": "release_hub", "release_id": 7, "section": "release_launchpad", "mode": "worker", }, secondary_focus_ref={ "kind": "release_hub", "release_id": 7, "section": "release_detail", }, ), ) ok, _, data = resolve_codex_action( { "entry_key": "runbook:release_progression", "requested_by": "test/codex-action", } ) self.assertTrue(ok) self.assertEqual("confirmation_required", data["gate"]["decision"]) self.assertFalse(data["gate"]["will_execute"]) self.assertTrue(data["gate"]["confirm_required"]) self.assertEqual("release_progression", data["gate"]["sequence_key"]) self.assertEqual("worker", data["preview_request"]["primary_focus_ref"]["mode"]) self.assertEqual("worker", data["gate"]["focus_ref"]["mode"]) self.assertEqual("worker", data["focus_ref"]["mode"]) @patch("app.services.ops_service.preview_driver_action") @patch("app.services.ops_service.get_ops_codex_brief") def test_resolve_codex_action_can_select_secondary_section( self, mock_get_ops_codex_brief, mock_preview_driver_action, ) -> None: mock_get_ops_codex_brief.return_value = { "focus": {"entry_key": "runbook:release_progression"}, "summary": {"total": 1}, "entries": [ _build_codex_brief_entry( key="runbook:release_progression", executor_kind="runbook_sequence", sequence_key="release_progression", primary_action_code="create_release_rollout_worker", recommendation="confirm_then_execute", automation_level="guarded_auto", target_api="/api/v1/ops/runbook/sequences/release_progression/execute", primary_action_payload={"release_id": 7}, secondary_action_code="focus_release_hub", secondary_action_payload={"release_id": 7}, focus_ref={ "kind": "release_hub", "release_id": 7, "section": "release_launchpad", }, primary_focus_ref={ "kind": "release_hub", "release_id": 7, "section": "release_launchpad", "mode": "worker", }, secondary_focus_ref={ "kind": "release_hub", "release_id": 7, "section": "release_detail", }, ) ], } preview_data = _build_preview_data( executor_kind="runbook_sequence", action_code="create_release_rollout_worker", recommendation="confirm_then_execute", automation_level="guarded_auto", target_api="/api/v1/ops/runbook/sequences/release_progression/execute", request_payload={ "secondary": False, "requested_by": "test/codex-action", "action_payload": { "release_id": 7, "focus_ref": { "kind": "release_hub", "release_id": 7, "section": "release_launchpad", "mode": "worker", }, }, }, sequence_key="release_progression", focus_ref={ "kind": "release_hub", "release_id": 7, "section": "release_launchpad", }, primary_focus_ref={ "kind": "release_hub", "release_id": 7, "section": "release_launchpad", "mode": "worker", }, secondary_focus_ref={ "kind": "release_hub", "release_id": 7, "section": "release_detail", }, ) preview_data["secondary"] = { "label": "查看版本区", "action_code": "focus_release_hub", "target_api": "/api/v1/ops/runbook/sequences/release_progression/execute", "target_method": "POST", "request_payload": { "secondary": True, "requested_by": "test/codex-action", "action_payload": { "release_id": 7, "focus_ref": { "kind": "release_hub", "release_id": 7, "section": "release_detail", }, }, }, "focus_ref": { "kind": "release_hub", "release_id": 7, "section": "release_detail", }, "execution_profile": { "action_code": "focus_release_hub", "backend_handled": False, "automation_level": "safe_auto", "recommendation": "auto_execute", "executor_mode_hint": "ui-intent", "risk_level": "low", "confirm_required": False, "ui_only": False, "blocked": False, "requires_fields": [], "missing_fields": [], "reason": "secondary preview reason", }, } mock_preview_driver_action.return_value = (True, "ok", preview_data) ok, _, data = resolve_codex_action( { "entry_key": "runbook:release_progression", "requested_by": "test/codex-action", "secondary": True, } ) self.assertTrue(ok) self.assertEqual("secondary", data["selected_section"]) self.assertTrue(data["gate"]["secondary"]) self.assertEqual("focus_release_hub", data["gate"]["action_code"]) self.assertEqual("auto_execute", data["gate"]["recommendation"]) self.assertEqual("release_detail", data["gate"]["focus_ref"]["section"]) self.assertEqual("release_detail", data["focus_ref"]["section"]) @patch("app.services.ops_service.preview_driver_action") @patch("app.services.ops_service.get_ops_codex_brief") def test_resolve_codex_action_can_fall_back_to_activity_focus_entry( self, mock_get_ops_codex_brief, mock_preview_driver_action, ) -> None: mock_get_ops_codex_brief.return_value = { "focus": {"entry_key": "activity:ops-job-42", "recommendation": "open_ui"}, "summary": {"total": 0, "activity_focus_total": 1}, "entries": [], "activity_focus": [ _build_codex_activity_focus_entry( key="activity:ops-job-42", activity_key="ops-job-42", activity_kind="ops_job", ui_intent_kind="job_events", focus_ref={ "kind": "ops_job", "job_key": "job-42", }, source_focus_ref={ "kind": "ops_job_event", "job_key": "job-42", "event_key": "event-1", }, target_node_codes=["mainland-controller-01"], ) ], } mock_preview_driver_action.return_value = ( True, "ok", _build_preview_data( executor_kind="driver_action", action_code="focus_activity_item", recommendation="auto_execute", automation_level="safe_auto", target_api="/api/v1/ops/driver-actions/execute", request_payload={ "action_code": "focus_activity_item", "node_codes": ["mainland-controller-01"], "action_payload": { "activity_key": "ops-job-42", "kind": "ops_job", "status": "attention", "ui_intent": { "kind": "job_events", "activity_key": "ops-job-42", }, "focus_ref": { "kind": "ops_job", "job_key": "job-42", }, }, "requested_by": "test/codex-action", }, focus_ref={ "kind": "ops_job", "job_key": "job-42", }, primary_focus_ref={ "kind": "ops_job", "job_key": "job-42", }, ), ) ok, _, data = resolve_codex_action({"requested_by": "test/codex-action"}) self.assertTrue(ok) self.assertEqual("activity:ops-job-42", data["entry_key"]) self.assertTrue(data["selected_entry"]["observation_only"]) self.assertEqual("ops_job", data["selected_entry"]["activity_kind"]) self.assertEqual("focus_activity_item", data["preview_request"]["primary_action_code"]) self.assertEqual("execute_now", data["gate"]["decision"]) self.assertTrue(data["gate"]["will_execute"]) self.assertEqual("job-42", data["focus_ref"]["job_key"]) @patch("app.services.ops_service.execute_driver_action") @patch("app.services.ops_service.preview_driver_action") @patch("app.services.ops_service.get_ops_codex_brief") def test_execute_codex_action_runs_safe_auto_driver_action( self, mock_get_ops_codex_brief, mock_preview_driver_action, mock_execute_driver_action, ) -> None: mock_get_ops_codex_brief.return_value = { "focus": {"entry_key": "recommendation:enable-log-sync"}, "summary": {"total": 1}, "entries": [ _build_codex_brief_entry( key="recommendation:enable-log-sync", executor_kind="driver_action", primary_action_code="enable_log_sync_key", recommendation="auto_execute", automation_level="safe_auto", target_api="/api/v1/ops/driver-actions/execute", ) ], } request_payload = { "action_code": "enable_log_sync_key", "node_codes": ["mainland-controller-01"], "action_payload": {}, "requested_by": "test/codex-action", } mock_preview_driver_action.return_value = ( True, "ok", _build_preview_data( executor_kind="driver_action", action_code="enable_log_sync_key", recommendation="auto_execute", automation_level="safe_auto", target_api="/api/v1/ops/driver-actions/execute", request_payload=request_payload, ), ) mock_execute_driver_action.return_value = ( True, "执行成功", {"handled": True, "action_code": "enable_log_sync_key"}, ) ok, message, data = execute_codex_action({"requested_by": "test/codex-action"}) self.assertTrue(ok) self.assertEqual("执行成功", message) self.assertTrue(data["executed"]) mock_execute_driver_action.assert_called_once_with(request_payload) @patch("app.services.ops_service.execute_ops_runbook_sequence") @patch("app.services.ops_service.preview_driver_action") @patch("app.services.ops_service.get_ops_codex_brief") def test_execute_codex_action_blocks_guarded_runbook_without_confirm( self, mock_get_ops_codex_brief, mock_preview_driver_action, mock_execute_ops_runbook_sequence, ) -> None: mock_get_ops_codex_brief.return_value = { "focus": {"entry_key": "runbook:release_progression"}, "summary": {"total": 1}, "entries": [ _build_codex_brief_entry( key="runbook:release_progression", executor_kind="runbook_sequence", sequence_key="release_progression", primary_action_code="create_release_rollout_worker", recommendation="confirm_then_execute", automation_level="guarded_auto", target_api="/api/v1/ops/runbook/sequences/release_progression/execute", primary_action_payload={"release_id": 7}, ) ], } mock_preview_driver_action.return_value = ( True, "ok", _build_preview_data( executor_kind="runbook_sequence", action_code="create_release_rollout_worker", recommendation="confirm_then_execute", automation_level="guarded_auto", target_api="/api/v1/ops/runbook/sequences/release_progression/execute", request_payload={ "secondary": False, "requested_by": "test/codex-action", "action_payload": {"release_id": 7}, }, sequence_key="release_progression", ), ) ok, message, data = execute_codex_action( { "entry_key": "runbook:release_progression", "requested_by": "test/codex-action", } ) self.assertFalse(ok) self.assertIn("确认", message) self.assertFalse(data["executed"]) mock_execute_ops_runbook_sequence.assert_not_called() @patch("app.services.ops_service.execute_driver_action") @patch("app.services.ops_service.preview_driver_action") @patch("app.services.ops_service.get_ops_codex_brief") def test_execute_codex_action_runs_activity_focus_as_safe_auto( self, mock_get_ops_codex_brief, mock_preview_driver_action, mock_execute_driver_action, ) -> None: mock_get_ops_codex_brief.return_value = { "focus": {"entry_key": "activity:ops-job-42", "recommendation": "open_ui"}, "summary": {"total": 0, "activity_focus_total": 1}, "entries": [], "activity_focus": [ _build_codex_activity_focus_entry( key="activity:ops-job-42", activity_key="ops-job-42", activity_kind="ops_job", ui_intent_kind="job_events", focus_ref={ "kind": "ops_job", "job_key": "job-42", }, ) ], } mock_preview_driver_action.return_value = ( True, "ok", _build_preview_data( executor_kind="driver_action", action_code="focus_activity_item", recommendation="auto_execute", automation_level="safe_auto", target_api="/api/v1/ops/driver-actions/execute", request_payload={ "action_code": "focus_activity_item", "node_codes": [], "action_payload": { "activity_key": "ops-job-42", "kind": "ops_job", "status": "attention", "ui_intent": { "kind": "job_events", "activity_key": "ops-job-42", }, "focus_ref": { "kind": "ops_job", "job_key": "job-42", }, }, "requested_by": "test/codex-action", }, focus_ref={ "kind": "ops_job", "job_key": "job-42", }, primary_focus_ref={ "kind": "ops_job", "job_key": "job-42", }, ), ) mock_execute_driver_action.return_value = ( True, "执行成功", {"handled": True, "action_code": "focus_activity_item"}, ) ok, message, data = execute_codex_action({"requested_by": "test/codex-action"}) self.assertTrue(ok) self.assertEqual("执行成功", message) self.assertTrue(data["executed"]) mock_execute_driver_action.assert_called_once() @patch("app.services.ops_service.preview_driver_action") @patch("app.services.ops_service.get_ops_codex_brief") def test_resolve_codex_action_supports_focus_latest_job_events_without_ui_intent( self, mock_get_ops_codex_brief, mock_preview_driver_action, ) -> None: mock_get_ops_codex_brief.return_value = { "focus": {"entry_key": "activity:ops-job-81", "recommendation": "open_ui"}, "summary": {"total": 0, "activity_focus_total": 1}, "entries": [], "activity_focus": [ { "key": "activity:ops-job-81", "activity_key": "ops-job-81", "activity_kind": "ops_job", "title": "开始回执异常", "subtitle": "ops-agent-81", "summary": "节点已接单执行，但开始回执未成功送达控制面。", "status": "attention", "status_label": "待处理", "meta_text": "ops-agent-81", "occurred_at": "2026-04-18T06:00:00", "target_node_codes": ["mainland-worker-01"], "focus_ref": { "kind": "ops_job", "job_id": 81, "job_code": "ops-agent-81", "target_node_code": "mainland-worker-01", }, "source_focus_ref": { "kind": "ops_job_event", "job_id": 81, "job_code": "ops-agent-81", "event_key": "job-event:81", "target_node_code": "mainland-worker-01", }, "ui_intent": {}, "ui_intent_kind": "", "focus_action_code": "focus_latest_job_events", "focus_action_payload": { "focus_ref": { "kind": "ops_job", "job_id": 81, "job_code": "ops-agent-81", "target_node_code": "mainland-worker-01", }, "source_focus_ref": { "kind": "ops_job_event", "job_id": 81, "job_code": "ops-agent-81", "event_key": "job-event:81", "target_node_code": "mainland-worker-01", }, }, "observation_only": True, } ], } mock_preview_driver_action.return_value = ( True, "ok", _build_preview_data( executor_kind="driver_action", action_code="focus_latest_job_events", recommendation="auto_execute", automation_level="safe_auto", target_api="/api/v1/ops/driver-actions/execute", request_payload={ "action_code": "focus_latest_job_events", "node_codes": ["mainland-worker-01"], "action_payload": { "focus_ref": { "kind": "ops_job", "job_id": 81, "job_code": "ops-agent-81", "target_node_code": "mainland-worker-01", }, "source_focus_ref": { "kind": "ops_job_event", "job_id": 81, "job_code": "ops-agent-81", "event_key": "job-event:81", "target_node_code": "mainland-worker-01", }, }, "requested_by": "test/codex-action", }, focus_ref={ "kind": "ops_job", "job_id": 81, "job_code": "ops-agent-81", }, primary_focus_ref={ "kind": "ops_job", "job_id": 81, "job_code": "ops-agent-81", }, ), ) ok, _, data = resolve_codex_action({"requested_by": "test/codex-action"}) self.assertTrue(ok) self.assertEqual("activity:ops-job-81", data["entry_key"]) self.assertEqual("focus_latest_job_events", data["preview_request"]["primary_action_code"]) self.assertEqual("execute_now", data["gate"]["decision"]) self.assertTrue(data["gate"]["will_execute"]) @patch("app.services.ops_service.execute_ops_runbook_sequence") @patch("app.services.ops_service.preview_driver_action") @patch("app.services.ops_service.get_ops_codex_brief") def test_execute_codex_action_runs_guarded_runbook_after_confirm( self, mock_get_ops_codex_brief, mock_preview_driver_action, mock_execute_ops_runbook_sequence, ) -> None: mock_get_ops_codex_brief.return_value = { "focus": {"entry_key": "runbook:release_progression"}, "summary": {"total": 1}, "entries": [ _build_codex_brief_entry( key="runbook:release_progression", executor_kind="runbook_sequence", sequence_key="release_progression", primary_action_code="create_release_rollout_worker", recommendation="confirm_then_execute", automation_level="guarded_auto", target_api="/api/v1/ops/runbook/sequences/release_progression/execute", primary_action_payload={"release_id": 7}, ) ], } request_payload = { "secondary": False, "requested_by": "test/codex-action", "action_payload": {"release_id": 7}, } mock_preview_driver_action.return_value = ( True, "ok", _build_preview_data( executor_kind="runbook_sequence", action_code="create_release_rollout_worker", recommendation="confirm_then_execute", automation_level="guarded_auto", target_api="/api/v1/ops/runbook/sequences/release_progression/execute", request_payload=request_payload, sequence_key="release_progression", ), ) mock_execute_ops_runbook_sequence.return_value = ( True, "runbook 执行成功", {"sequence_key": "release_progression", "driver_action_code": "create_release_rollout_worker"}, ) ok, message, data = execute_codex_action( { "entry_key": "runbook:release_progression", "requested_by": "test/codex-action", "confirm": True, } ) self.assertTrue(ok) self.assertEqual("runbook 执行成功", message) self.assertTrue(data["executed"]) mock_execute_ops_runbook_sequence.assert_called_once_with("release_progression", request_payload) if __name__ == "__main__": unittest.main()