498 lines
25 KiB
Python
498 lines
25 KiB
Python
import unittest
|
||
from unittest.mock import patch
|
||
|
||
from app.services.ops_service import get_ops_codex_brief
|
||
|
||
|
||
class OpsServiceCodexBriefTests(unittest.TestCase):
|
||
@patch("app.services.ops_service.get_ops_driver_feed")
|
||
def test_codex_brief_classifies_safe_ui_and_runbook_entries(self, mock_get_ops_driver_feed) -> None:
|
||
mock_get_ops_driver_feed.return_value = {
|
||
"headline": "当前建议已形成。",
|
||
"go_live_summary": {
|
||
"go_live_status": "attention",
|
||
"publish_ready": False,
|
||
"publish_status": "attention",
|
||
"publish_status_label": "可发布但建议先复核",
|
||
"operator_title": "先补现场日志回传",
|
||
"next_step_action_code": "enable_log_sync_key",
|
||
"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,
|
||
},
|
||
"activity_focus": [],
|
||
"entries": [
|
||
{
|
||
"key": "recommendation:enable-scene-log-sync",
|
||
"kind": "driver_recommendation",
|
||
"lane": "do_now",
|
||
"lane_label": "建议动作",
|
||
"title": "先补现场日志回传",
|
||
"summary": "优先打开关键日志回传。",
|
||
"reason": "当前还看不到现场过程日志。",
|
||
"status": "attention",
|
||
"level_label": "最高优先",
|
||
"tag_type": "danger",
|
||
"meta_text": "",
|
||
"detail_lines": ["参与节点 2 台"],
|
||
"node_codes": ["mainland-worker-01"],
|
||
"executor_kind": "driver_action",
|
||
"focus_ref": {
|
||
"kind": "execution_scene",
|
||
"scene_key": "scene.logs",
|
||
},
|
||
"primary_focus_ref": {
|
||
"kind": "execution_scene",
|
||
"scene_key": "scene.logs",
|
||
},
|
||
"secondary_focus_ref": {
|
||
"kind": "execution_scene",
|
||
"scene_key": "scene.logs.full",
|
||
},
|
||
"primary_label": "开启关键回传",
|
||
"secondary_label": "开启全量回传",
|
||
"primary_action_code": "enable_log_sync_key",
|
||
"secondary_action_code": "enable_log_sync_full",
|
||
"primary_node_codes": [],
|
||
"secondary_node_codes": [],
|
||
"primary_action_payload": {
|
||
"focus_ref": {
|
||
"kind": "execution_scene",
|
||
"scene_key": "scene.logs",
|
||
}
|
||
},
|
||
"secondary_action_payload": {
|
||
"focus_ref": {
|
||
"kind": "execution_scene",
|
||
"scene_key": "scene.logs.full",
|
||
}
|
||
},
|
||
},
|
||
{
|
||
"key": "recommendation:release-missing",
|
||
"kind": "driver_recommendation",
|
||
"lane": "do_now",
|
||
"lane_label": "建议动作",
|
||
"title": "先建立 Release",
|
||
"summary": "当前还没有默认 Release。",
|
||
"reason": "需要先进入版本区。",
|
||
"status": "planned",
|
||
"level_label": "待补齐",
|
||
"tag_type": "info",
|
||
"meta_text": "",
|
||
"detail_lines": [],
|
||
"node_codes": [],
|
||
"executor_kind": "driver_action",
|
||
"focus_ref": {
|
||
"kind": "release_hub",
|
||
"release_id": 7,
|
||
"section": "release_launchpad",
|
||
},
|
||
"primary_focus_ref": {
|
||
"kind": "release_hub",
|
||
"release_id": 7,
|
||
"section": "release_launchpad",
|
||
},
|
||
"primary_label": "创建 Release",
|
||
"secondary_label": "",
|
||
"primary_action_code": "open_release_dialog",
|
||
"secondary_action_code": "",
|
||
"primary_node_codes": [],
|
||
"secondary_node_codes": [],
|
||
"primary_action_payload": {},
|
||
"secondary_action_payload": {},
|
||
},
|
||
{
|
||
"key": "runbook:release_progression",
|
||
"kind": "runbook_sequence",
|
||
"lane": "standard_path",
|
||
"lane_label": "标准路径",
|
||
"title": "版本发布与放量",
|
||
"summary": "当前可以进入正式发布路径。",
|
||
"reason": "统一回到 Release / Rollout 闭环。",
|
||
"status": "attention",
|
||
"level_label": "待处理",
|
||
"tag_type": "warning",
|
||
"meta_text": "Release / Rollout",
|
||
"detail_lines": ["当前解析：Worker 灰度"],
|
||
"node_codes": ["mainland-controller-01"],
|
||
"executor_kind": "runbook_sequence",
|
||
"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",
|
||
},
|
||
"primary_label": "Worker 灰度",
|
||
"secondary_label": "查看版本区",
|
||
"primary_action_code": "create_release_rollout_worker",
|
||
"secondary_action_code": "focus_release_hub",
|
||
"primary_node_codes": ["mainland-controller-01"],
|
||
"secondary_node_codes": [],
|
||
"primary_action_payload": {
|
||
"release_id": 7,
|
||
"focus_ref": {
|
||
"kind": "release_hub",
|
||
"release_id": 7,
|
||
"section": "release_launchpad",
|
||
"mode": "worker",
|
||
},
|
||
},
|
||
"secondary_action_payload": {
|
||
"release_id": 7,
|
||
"focus_ref": {
|
||
"kind": "release_hub",
|
||
"release_id": 7,
|
||
"section": "release_detail",
|
||
},
|
||
},
|
||
"occurred_at": "2026-04-18T06:00:00",
|
||
},
|
||
],
|
||
}
|
||
|
||
payload = get_ops_codex_brief()
|
||
|
||
self.assertEqual(3, payload["summary"]["total"])
|
||
self.assertEqual("attention", payload["go_live_summary"]["go_live_status"])
|
||
self.assertEqual("attention", payload["summary"]["go_live_status"])
|
||
self.assertEqual("先补现场日志回传", payload["summary"]["operator_title"])
|
||
self.assertEqual("attention", payload["summary"]["publish_status"])
|
||
self.assertFalse(payload["summary"]["publish_ready"])
|
||
self.assertEqual("可发布但建议先复核", payload["summary"]["publish_status_label"])
|
||
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(0, payload["summary"]["activity_focus_total"])
|
||
self.assertEqual("当前建议已形成。", payload["driver_feed"]["headline"])
|
||
self.assertEqual("ops_driver_contract", payload["contract_navigation"]["primary_contract_key"])
|
||
self.assertIn("ops_stack_diagnosis_contract", payload["contract_navigation"]["contract_keys"])
|
||
self.assertEqual(2, payload["summary"]["auto_execute_total"])
|
||
self.assertEqual(1, payload["summary"]["confirm_then_execute_total"])
|
||
self.assertEqual(0, payload["summary"]["open_ui_total"])
|
||
self.assertEqual(1, payload["summary"]["preview_only_total"])
|
||
self.assertEqual(3, payload["summary"]["backend_handled_total"])
|
||
self.assertEqual("attention", payload["summary"]["launch_status"])
|
||
self.assertFalse(payload["summary"]["launch_ready"])
|
||
self.assertEqual("recommendation:enable-scene-log-sync", payload["focus"]["entry_key"])
|
||
self.assertEqual("execution_scene", payload["focus"]["focus_ref"]["kind"])
|
||
self.assertEqual("scene.logs", payload["focus"]["primary_focus_ref"]["scene_key"])
|
||
self.assertEqual("auto_execute", payload["entries"][0]["autopilot"]["recommendation"])
|
||
self.assertIn("ops_observability_contract", payload["entries"][0]["contract_keys"])
|
||
self.assertIn("ops_stack_diagnosis_contract", payload["entries"][0]["contract_keys"])
|
||
self.assertEqual("scene.logs", payload["entries"][0]["primary_focus_ref"]["scene_key"])
|
||
self.assertEqual("safe_auto", payload["entries"][1]["autopilot"]["automation_level"])
|
||
self.assertIn("release_hub_contract", payload["entries"][1]["contract_keys"])
|
||
|
||
runbook_entry = payload["entries"][2]
|
||
self.assertEqual("runbook_sequence", runbook_entry["executor_kind"])
|
||
self.assertEqual("confirm_then_execute", runbook_entry["autopilot"]["recommendation"])
|
||
self.assertIn("ops_playbook_contract", runbook_entry["contract_keys"])
|
||
self.assertIn("release_hub_contract", runbook_entry["contract_keys"])
|
||
self.assertEqual("release_launchpad", runbook_entry["focus_ref"]["section"])
|
||
self.assertEqual("worker", runbook_entry["primary_focus_ref"]["mode"])
|
||
self.assertEqual("release_detail", runbook_entry["secondary_focus_ref"]["section"])
|
||
self.assertEqual(
|
||
"/api/v1/ops/runbook/sequences/release_progression/execute",
|
||
runbook_entry["autopilot"]["target_api"],
|
||
)
|
||
self.assertEqual(
|
||
{
|
||
"sequence_key": "release_progression",
|
||
"secondary": False,
|
||
"action_payload": {
|
||
"release_id": 7,
|
||
"focus_ref": {
|
||
"kind": "release_hub",
|
||
"release_id": 7,
|
||
"section": "release_launchpad",
|
||
"mode": "worker",
|
||
},
|
||
},
|
||
},
|
||
runbook_entry["primary_execution"]["request_payload_preview"],
|
||
)
|
||
self.assertEqual(
|
||
{
|
||
"sequence_key": "release_progression",
|
||
"secondary": True,
|
||
"action_payload": {
|
||
"release_id": 7,
|
||
"focus_ref": {
|
||
"kind": "release_hub",
|
||
"release_id": 7,
|
||
"section": "release_detail",
|
||
},
|
||
},
|
||
},
|
||
runbook_entry["secondary_execution"]["request_payload_preview"],
|
||
)
|
||
self.assertEqual("attention", payload["automation_coverage"]["launch_status"])
|
||
self.assertEqual(2, payload["automation_coverage"]["safe_auto_total"])
|
||
self.assertEqual(1, payload["automation_coverage"]["guarded_auto_total"])
|
||
self.assertEqual(1, payload["automation_coverage"]["preview_only_total"])
|
||
self.assertEqual(3, payload["automation_coverage"]["backend_handled_total"])
|
||
|
||
@patch("app.services.ops_service.get_ops_driver_feed")
|
||
def test_codex_brief_marks_mixed_and_blocked_entries(self, mock_get_ops_driver_feed) -> None:
|
||
mock_get_ops_driver_feed.return_value = {
|
||
"headline": "当前建议需要先复核。",
|
||
"go_live_summary": {
|
||
"go_live_status": "blocked",
|
||
"publish_ready": False,
|
||
"publish_status": "blocked",
|
||
"publish_status_label": "暂不可发布",
|
||
"operator_title": "先补执行面接管",
|
||
"next_step_action_code": "fix_managed_nodes",
|
||
"launchpad_recommended_target_node_code": "mainland-controller-01",
|
||
"launchpad_recommended_recovery_label": "补执行面接管",
|
||
"launchpad_recommended_recovery_summary": "先处理 launchpad 缺口。",
|
||
"launchpad_onboarding_bootstrap_pending_nodes": 1,
|
||
"launchpad_onboarding_acceptance_ready_nodes": 0,
|
||
},
|
||
"activity_focus": [],
|
||
"entries": [
|
||
{
|
||
"key": "recommendation:release-launchpad-fix-managed",
|
||
"kind": "driver_recommendation",
|
||
"lane": "do_now",
|
||
"lane_label": "建议动作",
|
||
"title": "先补执行面接管",
|
||
"summary": "先处理 launchpad 缺口。",
|
||
"reason": "当前动作可能直接 sync，也可能转为补接入。",
|
||
"status": "attention",
|
||
"level_label": "最高优先",
|
||
"tag_type": "danger",
|
||
"meta_text": "",
|
||
"detail_lines": [],
|
||
"node_codes": ["mainland-controller-01"],
|
||
"executor_kind": "driver_action",
|
||
"focus_ref": {
|
||
"kind": "release_hub",
|
||
"release_id": 9,
|
||
"section": "release_launchpad",
|
||
},
|
||
"primary_focus_ref": {
|
||
"kind": "release_hub",
|
||
"release_id": 9,
|
||
"section": "default_rollout_gate",
|
||
},
|
||
"primary_label": "补执行面接管",
|
||
"secondary_label": "",
|
||
"primary_action_code": "fix_managed_nodes",
|
||
"secondary_action_code": "",
|
||
"primary_node_codes": ["mainland-controller-01"],
|
||
"secondary_node_codes": [],
|
||
"primary_action_payload": {"channel": "stable"},
|
||
"secondary_action_payload": {},
|
||
},
|
||
{
|
||
"key": "recommendation:inspect-participating",
|
||
"kind": "driver_recommendation",
|
||
"lane": "do_now",
|
||
"lane_label": "建议动作",
|
||
"title": "先看执行现场",
|
||
"summary": "先跑参与节点巡检。",
|
||
"reason": "缺少目标节点时不应误判为可自动执行。",
|
||
"status": "attention",
|
||
"level_label": "推荐",
|
||
"tag_type": "warning",
|
||
"meta_text": "",
|
||
"detail_lines": [],
|
||
"node_codes": [],
|
||
"executor_kind": "driver_action",
|
||
"primary_label": "执行标准巡检",
|
||
"secondary_label": "",
|
||
"primary_action_code": "run_inspection_participating",
|
||
"secondary_action_code": "",
|
||
"primary_node_codes": [],
|
||
"secondary_node_codes": [],
|
||
"primary_action_payload": {},
|
||
"secondary_action_payload": {},
|
||
},
|
||
],
|
||
}
|
||
|
||
payload = get_ops_codex_brief()
|
||
|
||
self.assertEqual("blocked", payload["summary"]["publish_status"])
|
||
self.assertEqual("暂不可发布", payload["summary"]["publish_status_label"])
|
||
self.assertEqual("mainland-controller-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(1, payload["summary"]["resolve_first_total"])
|
||
self.assertEqual(1, payload["summary"]["blocked_total"])
|
||
self.assertEqual("blocked", payload["summary"]["launch_status"])
|
||
self.assertFalse(payload["summary"]["launch_ready"])
|
||
self.assertEqual("mixed", payload["entries"][0]["autopilot"]["automation_level"])
|
||
self.assertEqual("resolve_first", payload["entries"][0]["autopilot"]["recommendation"])
|
||
self.assertEqual("default_rollout_gate", payload["entries"][0]["primary_focus_ref"]["section"])
|
||
self.assertEqual(1, payload["automation_coverage"]["mixed_total"])
|
||
self.assertEqual(1, payload["automation_coverage"]["blocked_total"])
|
||
self.assertEqual(2, payload["automation_coverage"]["human_dependency_total"])
|
||
|
||
blocked_entry = payload["entries"][1]
|
||
self.assertTrue(blocked_entry["autopilot"]["blocked"])
|
||
self.assertEqual(["node_codes"], blocked_entry["primary_execution"]["missing_fields"])
|
||
self.assertEqual("blocked", blocked_entry["autopilot"]["recommendation"])
|
||
|
||
@patch("app.services.ops_service.get_ops_driver_feed")
|
||
def test_codex_brief_exposes_activity_focus_and_uses_it_as_fallback_focus(self, mock_get_ops_driver_feed) -> None:
|
||
mock_get_ops_driver_feed.return_value = {
|
||
"headline": "当前先盯住执行现场。",
|
||
"scene_log_observation": {},
|
||
"entries": [],
|
||
"activity_focus": [
|
||
{
|
||
"key": "activity:execution-scene:live",
|
||
"activity_key": "execution-scene:live",
|
||
"activity_kind": "execution_scene",
|
||
"title": "当前执行现场",
|
||
"subtitle": "运行中",
|
||
"summary": "当前参与检测节点 2 台，建议先看现场与日志回传。",
|
||
"status": "running",
|
||
"status_label": "执行中",
|
||
"meta_text": "执行/已领 2 / 待命 1",
|
||
"occurred_at": "2026-04-18 09:30:00",
|
||
"target_node_codes": ["mainland-worker-01", "mainland-controller-01"],
|
||
"focus_ref": {
|
||
"kind": "execution_scene",
|
||
"scene_key": "scene.logs",
|
||
},
|
||
"source_focus_ref": {},
|
||
"ui_intent": {
|
||
"kind": "focus_execution_scene",
|
||
"open_log_drawer": True,
|
||
},
|
||
"ui_intent_kind": "focus_execution_scene",
|
||
"focus_action_code": "focus_activity_item",
|
||
"focus_action_payload": {
|
||
"activity_key": "execution-scene:live",
|
||
"kind": "execution_scene",
|
||
"status": "running",
|
||
"ui_intent": {
|
||
"kind": "focus_execution_scene",
|
||
"open_log_drawer": True,
|
||
},
|
||
"focus_ref": {
|
||
"kind": "execution_scene",
|
||
"scene_key": "scene.logs",
|
||
},
|
||
},
|
||
"observation_only": True,
|
||
}
|
||
],
|
||
}
|
||
|
||
payload = get_ops_codex_brief()
|
||
|
||
self.assertEqual(0, payload["summary"]["total"])
|
||
self.assertEqual(1, payload["summary"]["activity_focus_total"])
|
||
self.assertEqual("activity:execution-scene:live", payload["focus"]["entry_key"])
|
||
self.assertEqual("open_ui", payload["focus"]["recommendation"])
|
||
self.assertIn("ops_observability_contract", payload["focus"]["contract_keys"])
|
||
self.assertEqual("execution_scene", payload["focus"]["focus_ref"]["kind"])
|
||
self.assertEqual(1, len(payload["activity_focus"]))
|
||
self.assertEqual("focus_execution_scene", payload["activity_focus"][0]["ui_intent_kind"])
|
||
self.assertEqual(
|
||
"ops_driver_contract",
|
||
payload["activity_focus"][0]["contract_navigation"]["primary_contract_key"],
|
||
)
|
||
self.assertTrue(payload["activity_focus"][0]["observation_only"])
|
||
|
||
@patch("app.services.ops_service.get_ops_driver_feed")
|
||
def test_codex_brief_can_fallback_to_scene_log_observation_focus(self, mock_get_ops_driver_feed) -> None:
|
||
mock_get_ops_driver_feed.return_value = {
|
||
"headline": "先看现场日志。",
|
||
"entries": [],
|
||
"activity_focus": [],
|
||
"scene_log_observation": {
|
||
"status": "waiting_sample",
|
||
"status_label": "等待样本",
|
||
"summary": "日志回传已开启，但样本还没形成。",
|
||
"focus_ref": {
|
||
"kind": "node_scene_log",
|
||
"node_code": "mainland-worker-01",
|
||
"mode": "key",
|
||
"limit": 120,
|
||
},
|
||
"contract_keys": ["ops_observability_contract", "ops_stack_diagnosis_contract"],
|
||
},
|
||
}
|
||
|
||
payload = get_ops_codex_brief()
|
||
|
||
self.assertEqual("scene-log-observation", payload["focus"]["entry_key"])
|
||
self.assertEqual("open_ui", payload["focus"]["recommendation"])
|
||
self.assertEqual("node_scene_log", payload["focus"]["focus_ref"]["kind"])
|
||
self.assertEqual("mainland-worker-01", payload["focus"]["focus_ref"]["node_code"])
|
||
self.assertEqual("waiting_sample", payload["scene_log_observation"]["status"])
|
||
|
||
@patch("app.services.ops_service.get_ops_driver_feed")
|
||
def test_codex_brief_marks_bootstrap_run_as_agent_protocol_flow(self, mock_get_ops_driver_feed) -> None:
|
||
mock_get_ops_driver_feed.return_value = {
|
||
"headline": "先执行节点接入收口。",
|
||
"activity_focus": [],
|
||
"entries": [
|
||
{
|
||
"key": "priority:handover-gap-bootstrap_run",
|
||
"kind": "priority_recommendation",
|
||
"lane": "do_now",
|
||
"lane_label": "当前优先",
|
||
"title": "先补接管缺口",
|
||
"summary": "Launchpad 已明确给出首台节点的接入收口动作。",
|
||
"reason": "mainland-worker-01 还缺接入收口，先执行 bootstrap。",
|
||
"status": "attention",
|
||
"level_label": "最高优先",
|
||
"tag_type": "warning",
|
||
"meta_text": "overview.recommendation",
|
||
"detail_lines": ["当前已收敛到单节点接入动作。"],
|
||
"node_codes": ["mainland-worker-01"],
|
||
"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"},
|
||
"secondary_focus_ref": {"kind": "managed_node", "node_code": "mainland-worker-01", "section": "bootstrap_plan"},
|
||
"primary_label": "跑接入收口",
|
||
"secondary_label": "查看 Bootstrap 计划",
|
||
"primary_action_code": "bootstrap_run",
|
||
"secondary_action_code": "create_bootstrap_plan",
|
||
"primary_node_codes": ["mainland-worker-01"],
|
||
"secondary_node_codes": ["mainland-worker-01"],
|
||
"primary_action_payload": {"node_code": "mainland-worker-01"},
|
||
"secondary_action_payload": {"node_code": "mainland-worker-01"},
|
||
}
|
||
],
|
||
}
|
||
|
||
payload = get_ops_codex_brief()
|
||
|
||
self.assertEqual("priority:handover-gap-bootstrap_run", payload["focus"]["entry_key"])
|
||
self.assertEqual("resolve_first", payload["focus"]["recommendation"])
|
||
self.assertIn("ops_agent_protocol", payload["entries"][0]["contract_keys"])
|
||
self.assertEqual("bootstrap_run", payload["entries"][0]["primary_action_code"])
|
||
self.assertEqual("mainland-worker-01", payload["entries"][0]["primary_execution"]["node_codes"][0])
|
||
self.assertEqual("/api/v1/ops/driver-actions/execute", payload["entries"][0]["autopilot"]["target_api"])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|