275 lines
12 KiB
Python
275 lines
12 KiB
Python
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()
|