1266 lines
50 KiB
Python
1266 lines
50 KiB
Python
import unittest
|
||
from unittest.mock import patch
|
||
|
||
from app.services.ops_service import execute_driver_action
|
||
|
||
|
||
class OpsServiceDriverActionTests(unittest.TestCase):
|
||
@patch("app.services.ops_service.execute_managed_node_onboarding_recovery")
|
||
def test_bootstrap_run_executes_node_onboarding_recovery(
|
||
self,
|
||
mock_execute_managed_node_onboarding_recovery,
|
||
) -> None:
|
||
mock_execute_managed_node_onboarding_recovery.return_value = (
|
||
True,
|
||
"自动收口任务已创建",
|
||
{
|
||
"recovery_action": "bootstrap_run",
|
||
"playbook_run": {"run_code": "run-001"},
|
||
},
|
||
)
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "bootstrap_run",
|
||
"node_codes": ["mainland-worker-02"],
|
||
"action_payload": {
|
||
"control_plane_base_url": "http://control.example:8100",
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("自动收口任务已创建", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("node-onboarding-recovery", data["mode"])
|
||
self.assertEqual("mainland-worker-02", data["node_code"])
|
||
mock_execute_managed_node_onboarding_recovery.assert_called_once_with(
|
||
node_code="mainland-worker-02",
|
||
requested_by="test/bootstrap_run",
|
||
control_plane_base_url="http://control.example:8100",
|
||
root_dir="",
|
||
)
|
||
|
||
@patch("app.services.ops_service.create_release_and_smart_rollout_from_latest_package")
|
||
def test_publish_latest_worker_calls_backend_launchpad_flow(
|
||
self,
|
||
mock_create_release_and_smart_rollout_from_latest_package,
|
||
) -> None:
|
||
mock_create_release_and_smart_rollout_from_latest_package.return_value = (
|
||
True,
|
||
"Release rollout 已创建",
|
||
{
|
||
"release": {"id": 51, "release_version": "domaincheck_release_launchpad"},
|
||
"rollout_created": True,
|
||
},
|
||
)
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "publish_latest_worker",
|
||
"action_payload": {
|
||
"channel": "stable",
|
||
"control_plane_base_url": "http://control.example:8100",
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("Release rollout 已创建", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("smart-release-rollout", data["mode"])
|
||
mock_create_release_and_smart_rollout_from_latest_package.assert_called_once()
|
||
called_payload = mock_create_release_and_smart_rollout_from_latest_package.call_args.args[0]
|
||
called_kwargs = mock_create_release_and_smart_rollout_from_latest_package.call_args.kwargs
|
||
self.assertEqual("worker", called_payload["mode"])
|
||
self.assertEqual("stable", called_payload["channel"])
|
||
self.assertEqual("test/publish_latest_worker", called_payload["created_by"])
|
||
self.assertEqual("test/publish_latest_worker/rollout", called_payload["rollout_created_by"])
|
||
self.assertEqual("http://control.example:8100", called_kwargs["control_plane_base_url"])
|
||
|
||
@patch("app.services.ops_service.prepare_latest_release_package")
|
||
def test_release_prepare_executes_final_signoff_flow(
|
||
self,
|
||
mock_prepare_latest_release_package,
|
||
) -> None:
|
||
mock_prepare_latest_release_package.return_value = (
|
||
True,
|
||
"最新发布包最终签收已完成",
|
||
{
|
||
"package": {
|
||
"available": True,
|
||
"package_name": "domaincheck_release_20260418_220000",
|
||
"final_release_ok": True,
|
||
},
|
||
"final_release_gate": {
|
||
"ready": True,
|
||
"status": "ready",
|
||
},
|
||
},
|
||
)
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "release_prepare",
|
||
"action_payload": {},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("最新发布包最终签收已完成", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("release-prepare", data["mode"])
|
||
self.assertTrue(data["release_prepare_result"]["package"]["final_release_ok"])
|
||
mock_prepare_latest_release_package.assert_called_once_with(
|
||
requested_by="test/release_prepare"
|
||
)
|
||
|
||
@patch("app.services.ops_service.get_runtime_build_info")
|
||
def test_api_restart_returns_runtime_refresh_preview(
|
||
self,
|
||
mock_get_runtime_build_info,
|
||
) -> None:
|
||
mock_get_runtime_build_info.return_value = {
|
||
"commit_sha": "246838ae4c07",
|
||
"route_surface": {
|
||
"surface_complete": False,
|
||
"missing_keys": ["ops_node_handover_bootstrap_plan"],
|
||
},
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "api-restart",
|
||
"action_payload": {
|
||
"control_plane_base_url": "http://control.example:8100",
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertIn("刷新控制面运行时", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("runtime-refresh-preview", data["mode"])
|
||
self.assertEqual("open_stack_diagnosis", data["ui_intent"]["kind"])
|
||
self.assertIn("runtime-refresh-recover", data["recommended_command"])
|
||
self.assertEqual(
|
||
["ops_node_handover_bootstrap_plan"],
|
||
data["runtime_build_info"]["route_surface"]["missing_keys"],
|
||
)
|
||
|
||
@patch("app.services.ops_service.get_release_launchpad")
|
||
def test_review_smart_rollout_preview_returns_launchpad_preview_payload(self, mock_get_release_launchpad) -> None:
|
||
mock_get_release_launchpad.return_value = {
|
||
"launchpad_status": {
|
||
"status": "attention",
|
||
"status_label": "待确认",
|
||
},
|
||
"latest_release": {
|
||
"id": 88,
|
||
"release_version": "domaincheck_release_demo",
|
||
},
|
||
"worker_rollout_preview": {
|
||
"message": "Worker 智能预案待确认",
|
||
"smart_rollout": {
|
||
"available": True,
|
||
"target_node_codes": ["mainland-worker-01"],
|
||
},
|
||
},
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "review_smart_rollout_preview",
|
||
"action_payload": {
|
||
"channel": "stable",
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("release-launchpad-preview", data["mode"])
|
||
self.assertEqual("worker", data["review_mode"])
|
||
self.assertEqual("worker", data["release_launchpad_review"]["mode"])
|
||
self.assertEqual(88, data["release_launchpad_review"]["latest_release"]["id"])
|
||
self.assertEqual(
|
||
["mainland-worker-01"],
|
||
data["release_launchpad_review"]["target_node_codes"],
|
||
)
|
||
self.assertEqual("release_launchpad_review", data["ui_intent"]["kind"])
|
||
self.assertEqual("worker", data["ui_intent"]["mode"])
|
||
self.assertEqual(88, data["ui_intent"]["release_id"])
|
||
mock_get_release_launchpad.assert_called_once()
|
||
|
||
@patch("app.services.ops_service.get_managed_node_handover")
|
||
@patch("app.services.ops_service.get_release_launchpad")
|
||
def test_fix_managed_nodes_returns_handover_ui_intent_when_gap_exists(
|
||
self,
|
||
mock_get_release_launchpad,
|
||
mock_get_managed_node_handover,
|
||
) -> None:
|
||
mock_get_release_launchpad.return_value = {
|
||
"worker_rollout_preview": {
|
||
"policy_preview": {
|
||
"release_gate": {
|
||
"operational_readiness": {
|
||
"rows": [
|
||
{
|
||
"node_code": "mainland-worker-01",
|
||
"is_managed": False,
|
||
"is_enabled": False,
|
||
"is_agent_online": False,
|
||
"remote_agent_ready": False,
|
||
"inspection_status": "missing",
|
||
}
|
||
]
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
mock_get_managed_node_handover.return_value = {
|
||
"node_code": "mainland-worker-01",
|
||
"stage": {"code": "unmanaged"},
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "fix_managed_nodes",
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("ui-intent", data["mode"])
|
||
self.assertEqual("managed_node_handover", data["ui_intent"]["kind"])
|
||
self.assertEqual("mainland-worker-01", data["ui_intent"]["node_code"])
|
||
self.assertEqual("unmanaged", data["handover"]["stage"]["code"])
|
||
|
||
@patch("app.services.ops_service.sync_managed_nodes_from_cluster")
|
||
@patch("app.services.ops_service.get_release_launchpad")
|
||
def test_fix_managed_nodes_syncs_cluster_when_gap_not_found(
|
||
self,
|
||
mock_get_release_launchpad,
|
||
mock_sync_managed_nodes_from_cluster,
|
||
) -> None:
|
||
mock_get_release_launchpad.return_value = {
|
||
"worker_rollout_preview": {
|
||
"policy_preview": {
|
||
"release_gate": {
|
||
"operational_readiness": {
|
||
"rows": []
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
mock_sync_managed_nodes_from_cluster.return_value = {
|
||
"synced_count": 2,
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "fix_managed_nodes",
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("已按集群快照同步托管节点，请继续检查接管状态。", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("cluster-sync", data["mode"])
|
||
self.assertEqual(2, data["sync_result"]["synced_count"])
|
||
mock_sync_managed_nodes_from_cluster.assert_called_once_with(dry_run=False)
|
||
|
||
@patch("app.services.ops_service.get_managed_node_handover")
|
||
def test_handover_first_gap_includes_handover_payload(self, mock_get_managed_node_handover) -> None:
|
||
mock_get_managed_node_handover.return_value = {
|
||
"node_code": "mainland-worker-01",
|
||
"stage": {"code": "pending_bootstrap"},
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "handover_first_gap",
|
||
"action_payload": {
|
||
"node_code": "mainland-worker-01",
|
||
"control_plane_base_url": "http://control.example:8100",
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("", message)
|
||
self.assertEqual("handover-preview", data["mode"])
|
||
self.assertEqual("pending_bootstrap", data["handover_preview"]["stage"]["code"])
|
||
self.assertEqual("managed_node_handover", data["ui_intent"]["kind"])
|
||
self.assertEqual("pending_bootstrap", data["handover"]["stage"]["code"])
|
||
mock_get_managed_node_handover.assert_called_once_with(
|
||
"mainland-worker-01",
|
||
control_plane_base_url="http://control.example:8100",
|
||
)
|
||
|
||
@patch("app.services.ops_service.get_managed_node_handover")
|
||
def test_view_first_gap_returns_handover_preview_and_edit_ui_intent(self, mock_get_managed_node_handover) -> None:
|
||
mock_get_managed_node_handover.return_value = {
|
||
"node_code": "mainland-worker-02",
|
||
"stage": {"code": "runtime_only"},
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "view_first_gap",
|
||
"action_payload": {
|
||
"node_code": "mainland-worker-02",
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("", message)
|
||
self.assertEqual("handover-preview", data["mode"])
|
||
self.assertEqual("runtime_only", data["handover_preview"]["stage"]["code"])
|
||
self.assertEqual("managed_node_edit", data["ui_intent"]["kind"])
|
||
self.assertEqual("runtime_only", data["handover"]["stage"]["code"])
|
||
|
||
@patch("app.services.ops_service.get_latest_release_package_metadata")
|
||
def test_release_package_returns_package_preview_and_open_release_dialog_ui_intent(
|
||
self,
|
||
mock_get_latest_release_package_metadata,
|
||
) -> None:
|
||
mock_get_latest_release_package_metadata.return_value = {
|
||
"available": True,
|
||
"package_name": "domaincheck_release_20260418_120000",
|
||
"release_version_suggestion": "domaincheck_release_20260418_120000",
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "release_package",
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("release-package-preview", data["mode"])
|
||
self.assertTrue(data["release_package_preview"]["available"])
|
||
self.assertEqual(
|
||
"domaincheck_release_20260418_120000",
|
||
data["release_package_preview"]["package_name"],
|
||
)
|
||
self.assertEqual("open_release_dialog", data["ui_intent"]["kind"])
|
||
mock_get_latest_release_package_metadata.assert_called_once_with()
|
||
|
||
@patch("app.services.ops_service.get_release_summary")
|
||
@patch("app.services.ops_service.get_latest_release_package_metadata")
|
||
def test_open_release_dialog_returns_release_dialog_preview_and_ui_intent(
|
||
self,
|
||
mock_get_latest_release_package_metadata,
|
||
mock_get_release_summary,
|
||
) -> None:
|
||
mock_get_latest_release_package_metadata.return_value = {
|
||
"available": True,
|
||
"package_name": "domaincheck_release_20260418_120000",
|
||
}
|
||
mock_get_release_summary.return_value = {
|
||
"latest_release": {
|
||
"id": 7,
|
||
"release_version": "2026.04.18",
|
||
}
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "open_release_dialog",
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("release-dialog-preview", data["mode"])
|
||
self.assertEqual(7, data["release_dialog_preview"]["summary"]["latest_release"]["id"])
|
||
self.assertEqual(
|
||
"domaincheck_release_20260418_120000",
|
||
data["release_dialog_preview"]["release_package_preview"]["package_name"],
|
||
)
|
||
self.assertEqual("open_release_dialog", data["ui_intent"]["kind"])
|
||
mock_get_latest_release_package_metadata.assert_called_once_with()
|
||
mock_get_release_summary.assert_called_once_with()
|
||
|
||
@patch("app.services.ops_service.get_release_launchpad")
|
||
@patch("app.services.ops_service.get_release_summary")
|
||
def test_focus_release_hub_returns_release_hub_preview_and_ui_intent(
|
||
self,
|
||
mock_get_release_summary,
|
||
mock_get_release_launchpad,
|
||
) -> None:
|
||
mock_get_release_summary.return_value = {
|
||
"latest_release": {
|
||
"id": 7,
|
||
"release_version": "2026.04.18",
|
||
}
|
||
}
|
||
mock_get_release_launchpad.return_value = {
|
||
"channel": "stable",
|
||
"launchpad_status": {
|
||
"status": "attention",
|
||
"recommended_action_code": "review_smart_rollout_preview",
|
||
},
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "focus_release_hub",
|
||
"action_payload": {
|
||
"release_id": 7,
|
||
"channel": "stable",
|
||
"control_plane_base_url": "http://control.example:8100",
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("release-hub-preview", data["mode"])
|
||
self.assertEqual("focus_release_hub", data["ui_intent"]["kind"])
|
||
self.assertEqual(7, data["ui_intent"]["release_id"])
|
||
self.assertEqual(
|
||
"review_smart_rollout_preview",
|
||
data["release_hub_preview"]["launchpad"]["launchpad_status"]["recommended_action_code"],
|
||
)
|
||
self.assertEqual(7, data["release_hub_preview"]["summary"]["latest_release"]["id"])
|
||
mock_get_release_summary.assert_called_once_with()
|
||
mock_get_release_launchpad.assert_called_once_with(
|
||
control_plane_base_url="http://control.example:8100",
|
||
channel="stable",
|
||
)
|
||
|
||
def test_open_playbook_dialog_returns_ui_intent(self) -> None:
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "open_playbook_dialog",
|
||
"node_codes": ["mainland-worker-01"],
|
||
"action_payload": {
|
||
"playbook_key": "scene.logs.key",
|
||
"execution_mode": "remote-agent",
|
||
"auto_approve": True,
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("playbook-preview", data["mode"])
|
||
self.assertEqual("scene.logs.key", data["playbook_preview"]["playbook"]["key"])
|
||
self.assertEqual(1, data["playbook_preview"]["target_nodes_total"])
|
||
self.assertEqual(["mainland-worker-01"], data["playbook_preview"]["target_node_codes"])
|
||
self.assertEqual("remote-agent", data["playbook_preview"]["execution_mode"])
|
||
self.assertTrue(data["playbook_preview"]["auto_approve"])
|
||
self.assertEqual("open_playbook_dialog", data["ui_intent"]["kind"])
|
||
self.assertEqual("scene.logs.key", data["ui_intent"]["playbook_key"])
|
||
self.assertEqual(["mainland-worker-01"], data["ui_intent"]["target_node_codes"])
|
||
self.assertEqual("remote-agent", data["ui_intent"]["execution_mode"])
|
||
self.assertTrue(data["ui_intent"]["auto_approve"])
|
||
|
||
@patch("app.services.ops_service.list_ops_playbook_run_events")
|
||
@patch("app.services.ops_service.get_ops_playbook_run")
|
||
def test_focus_playbook_run_returns_run_focus_preview_and_ui_intent(
|
||
self,
|
||
mock_get_ops_playbook_run,
|
||
mock_list_ops_playbook_run_events,
|
||
) -> None:
|
||
mock_get_ops_playbook_run.return_value = {
|
||
"run_code": "pbr-demo",
|
||
"playbook_key": "inspection.standard",
|
||
"status": "running",
|
||
}
|
||
mock_list_ops_playbook_run_events.return_value = {
|
||
"run_code": "pbr-demo",
|
||
"events": [{"event_key": "evt-1"}],
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "focus_playbook_run",
|
||
"action_payload": {
|
||
"run_code": "pbr-demo",
|
||
"focus_step_key": "logs.collect",
|
||
"focus_step_title": "收集服务日志",
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("playbook-run-focus", data["mode"])
|
||
self.assertEqual("pbr-demo", data["playbook_run_focus"]["run"]["run_code"])
|
||
self.assertEqual("evt-1", data["playbook_run_focus"]["events"]["events"][0]["event_key"])
|
||
self.assertEqual("playbook_run_detail", data["ui_intent"]["kind"])
|
||
mock_get_ops_playbook_run.assert_called_once_with("pbr-demo")
|
||
mock_list_ops_playbook_run_events.assert_called_once_with(
|
||
"pbr-demo",
|
||
limit=40,
|
||
step_key="logs.collect",
|
||
node_code="",
|
||
)
|
||
|
||
@patch("app.services.ops_service.get_ops_activity_stream")
|
||
def test_focus_activity_item_returns_activity_focus_preview_and_ui_intent(
|
||
self,
|
||
mock_get_ops_activity_stream,
|
||
) -> None:
|
||
mock_get_ops_activity_stream.return_value = {
|
||
"items": [
|
||
{
|
||
"activity_key": "ops-job:42",
|
||
"kind": "ops_job",
|
||
"status": "attention",
|
||
"summary": "任务待关注",
|
||
}
|
||
]
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "focus_activity_item",
|
||
"action_payload": {
|
||
"activity_key": "ops-job:42",
|
||
"kind": "ops_job",
|
||
"status": "attention",
|
||
"ui_intent": {
|
||
"kind": "job_events",
|
||
"job_id": 42,
|
||
},
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("activity-focus", data["mode"])
|
||
self.assertEqual("ops-job:42", data["activity_focus_preview"]["activity"]["activity_key"])
|
||
self.assertEqual("job_events", data["ui_intent"]["kind"])
|
||
|
||
@patch("app.services.ops_service.list_ops_job_events_for_jobs")
|
||
def test_focus_latest_job_events_returns_job_events_preview_and_ui_intent(
|
||
self,
|
||
mock_list_ops_job_events_for_jobs,
|
||
) -> None:
|
||
mock_list_ops_job_events_for_jobs.return_value = [
|
||
{
|
||
"id": 1,
|
||
"job_id": 42,
|
||
"event_type": "job_started",
|
||
"message": "started",
|
||
}
|
||
]
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "focus_latest_job_events",
|
||
"action_payload": {
|
||
"focus_ref": {
|
||
"kind": "ops_job",
|
||
"job_id": 42,
|
||
"job_code": "job-42",
|
||
}
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("job-events-focus", data["mode"])
|
||
self.assertEqual(42, data["job_events_preview"]["job_id"])
|
||
self.assertEqual("job-42", data["job_events_preview"]["job_code"])
|
||
self.assertEqual(1, data["job_events_preview"]["events"][0]["id"])
|
||
self.assertEqual("job_events", data["ui_intent"]["kind"])
|
||
mock_list_ops_job_events_for_jobs.assert_called_once_with([42], limit=60)
|
||
|
||
@patch("app.services.ops_service.list_ops_playbook_run_events")
|
||
def test_open_playbook_run_latest_events_returns_events_preview_and_ui_intent(
|
||
self,
|
||
mock_list_ops_playbook_run_events,
|
||
) -> None:
|
||
mock_list_ops_playbook_run_events.return_value = {
|
||
"run_code": "pbr-demo",
|
||
"events": [{"event_key": "evt-2"}],
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "open_playbook_run_latest_events",
|
||
"action_payload": {
|
||
"run_code": "pbr-demo",
|
||
"focus_step_key": "logs.collect",
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("playbook-run-events", data["mode"])
|
||
self.assertEqual("evt-2", data["playbook_run_events_preview"]["events"][0]["event_key"])
|
||
self.assertEqual("playbook_run_latest_events", data["ui_intent"]["kind"])
|
||
mock_list_ops_playbook_run_events.assert_called_once_with(
|
||
"pbr-demo",
|
||
limit=60,
|
||
step_key="logs.collect",
|
||
node_code="",
|
||
)
|
||
|
||
@patch("app.services.ops_service.list_managed_nodes_with_agent_state")
|
||
def test_open_playbook_dialog_auto_switches_to_ssh_when_supported_and_target_is_ssh_ready(
|
||
self,
|
||
mock_list_managed_nodes_with_agent_state,
|
||
) -> None:
|
||
mock_list_managed_nodes_with_agent_state.return_value = {
|
||
"nodes": [
|
||
{
|
||
"node_code": "mainland-worker-01",
|
||
"is_managed": True,
|
||
"is_enabled": True,
|
||
"is_agent_online": False,
|
||
"has_ssh_access": True,
|
||
}
|
||
]
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "open_playbook_dialog",
|
||
"node_codes": ["mainland-worker-01"],
|
||
"action_payload": {
|
||
"playbook_key": "scene.logs.key",
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("ssh", data["ui_intent"]["execution_mode"])
|
||
|
||
def test_open_playbook_dialog_keeps_control_plane_when_playbook_only_supports_control_plane(self) -> None:
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "open_playbook_dialog",
|
||
"node_codes": ["mainland-worker-01"],
|
||
"action_payload": {
|
||
"playbook_key": "onboarding.bootstrap",
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("control-plane", data["ui_intent"]["execution_mode"])
|
||
|
||
def test_open_action_template_dialog_returns_ui_intent(self) -> None:
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "open_action_template_dialog",
|
||
"node_codes": ["mainland-controller-01"],
|
||
"action_payload": {
|
||
"template_key": "logs.collect",
|
||
"execution_mode": "remote-agent",
|
||
"auto_approve": False,
|
||
"payload": {
|
||
"service_name": "domaincheck-worker",
|
||
"lines": 120,
|
||
},
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("action-template-preview", data["mode"])
|
||
self.assertEqual("logs.collect", data["template_preview"]["template"]["key"])
|
||
self.assertEqual("remote-agent", data["template_preview"]["execution_mode"])
|
||
self.assertEqual(["mainland-controller-01"], data["template_preview"]["target_node_codes"])
|
||
self.assertEqual("domaincheck-worker", data["template_preview"]["payload"]["service_name"])
|
||
self.assertEqual("open_action_template_dialog", data["ui_intent"]["kind"])
|
||
self.assertEqual("logs.collect", data["ui_intent"]["template_key"])
|
||
self.assertEqual(["mainland-controller-01"], data["ui_intent"]["target_node_codes"])
|
||
self.assertEqual("remote-agent", data["ui_intent"]["execution_mode"])
|
||
self.assertFalse(data["ui_intent"]["auto_approve"])
|
||
self.assertEqual("domaincheck-worker", data["ui_intent"]["payload"]["service_name"])
|
||
self.assertEqual(120, data["ui_intent"]["payload"]["lines"])
|
||
|
||
@patch("app.services.ops_service.create_ops_job_batch")
|
||
def test_replay_delivery_queue_creates_ops_job_batch(self, mock_create_ops_job_batch) -> None:
|
||
mock_create_ops_job_batch.return_value = (
|
||
True,
|
||
"已创建 2 个运维任务；待审批 0；阻断 0；失败 0",
|
||
{
|
||
"summary": {"created_count": 2},
|
||
"jobs": [{"id": 1}, {"id": 2}],
|
||
},
|
||
)
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "replay_delivery_queue",
|
||
"node_codes": ["mainland-worker-01", "mainland-worker-02"],
|
||
"action_payload": {
|
||
"payload": {
|
||
"limit": 2,
|
||
"flush_after_replay": True,
|
||
},
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("已创建 2 个运维任务；待审批 0；阻断 0；失败 0", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("ops-job-batch", data["mode"])
|
||
self.assertEqual("delivery.queue.replay", data["template_key"])
|
||
called_payload = mock_create_ops_job_batch.call_args.args[0]
|
||
self.assertEqual("delivery.queue.replay", called_payload["template_key"])
|
||
self.assertEqual(["mainland-worker-01", "mainland-worker-02"], called_payload["target_node_codes"])
|
||
self.assertEqual("test/replay_delivery_queue", called_payload["requested_by"])
|
||
self.assertEqual("remote-agent", called_payload["execution_mode"])
|
||
self.assertFalse(called_payload["auto_approve"])
|
||
self.assertTrue(called_payload["payload"]["flush_after_replay"])
|
||
|
||
@patch("app.services.ops_service.create_ops_job_batch")
|
||
def test_flush_delivery_queue_creates_ops_job_batch(self, mock_create_ops_job_batch) -> None:
|
||
mock_create_ops_job_batch.return_value = (
|
||
True,
|
||
"已创建 1 个运维任务；待审批 0；阻断 0；失败 0",
|
||
{
|
||
"summary": {"created_count": 1},
|
||
"jobs": [{"id": 9}],
|
||
},
|
||
)
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "flush_delivery_queue",
|
||
"action_payload": {
|
||
"target_node_codes": ["mainland-controller-01"],
|
||
"auto_approve": True,
|
||
"payload": {
|
||
"limit": 5,
|
||
},
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("ops-job-batch", data["mode"])
|
||
self.assertEqual("delivery.queue.flush", data["template_key"])
|
||
called_payload = mock_create_ops_job_batch.call_args.args[0]
|
||
self.assertEqual("delivery.queue.flush", called_payload["template_key"])
|
||
self.assertEqual(["mainland-controller-01"], called_payload["target_node_codes"])
|
||
self.assertEqual("test/flush_delivery_queue", called_payload["requested_by"])
|
||
self.assertTrue(called_payload["auto_approve"])
|
||
self.assertEqual(5, called_payload["payload"]["limit"])
|
||
|
||
@patch("app.services.ops_service.list_managed_nodes_with_agent_state")
|
||
def test_open_action_template_dialog_auto_switches_to_ssh_when_supported_and_target_is_ssh_ready(
|
||
self,
|
||
mock_list_managed_nodes_with_agent_state,
|
||
) -> None:
|
||
mock_list_managed_nodes_with_agent_state.return_value = {
|
||
"nodes": [
|
||
{
|
||
"node_code": "mainland-controller-01",
|
||
"is_managed": True,
|
||
"is_enabled": True,
|
||
"is_agent_online": False,
|
||
"has_ssh_access": True,
|
||
}
|
||
]
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "open_action_template_dialog",
|
||
"node_codes": ["mainland-controller-01"],
|
||
"action_payload": {
|
||
"template_key": "logs.collect",
|
||
"payload": {
|
||
"service_name": "domaincheck-api",
|
||
"lines": 120,
|
||
},
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("ssh", data["ui_intent"]["execution_mode"])
|
||
|
||
@patch("app.services.ops_service.execute_ops_playbook")
|
||
def test_run_scene_logs_full_executes_ops_playbook(
|
||
self,
|
||
mock_execute_ops_playbook,
|
||
) -> None:
|
||
mock_execute_ops_playbook.return_value = (
|
||
True,
|
||
"现场取证已创建",
|
||
{
|
||
"run_code": "playbook-logs-full-001",
|
||
},
|
||
)
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "run_scene_logs_full",
|
||
"node_codes": ["mainland-worker-01"],
|
||
"action_payload": {
|
||
"execution_mode": "remote-agent",
|
||
"auto_approve": True,
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("现场取证已创建", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("ops-playbook", data["mode"])
|
||
self.assertEqual("scene.logs.full", data["playbook_key"])
|
||
called_payload = mock_execute_ops_playbook.call_args.args[0]
|
||
self.assertEqual("scene.logs.full", called_payload["playbook_key"])
|
||
self.assertEqual(["mainland-worker-01"], called_payload["node_codes"])
|
||
self.assertEqual("remote-agent", called_payload["execution_mode"])
|
||
self.assertTrue(called_payload["auto_approve"])
|
||
self.assertEqual("test/run_scene_logs_full", called_payload["requested_by"])
|
||
|
||
@patch("app.services.ops_service.get_release_launchpad")
|
||
@patch("app.services.ops_service.get_release_summary")
|
||
def test_open_release_deploy_worker_returns_preview_and_ui_intent(
|
||
self,
|
||
mock_get_release_summary,
|
||
mock_get_release_launchpad,
|
||
) -> None:
|
||
mock_get_release_summary.return_value = {
|
||
"latest_release": {
|
||
"id": 11,
|
||
"release_version": "2026.04.18",
|
||
}
|
||
}
|
||
mock_get_release_launchpad.return_value = {
|
||
"launchpad_status": {
|
||
"status": "attention",
|
||
"recommended_action_code": "review_smart_rollout_preview",
|
||
}
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "open_release_deploy_worker",
|
||
"node_codes": ["mainland-worker-01"],
|
||
"action_payload": {
|
||
"release_id": 11,
|
||
"execution_mode": "remote-agent",
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("release-deploy-preview", data["mode"])
|
||
self.assertEqual("deploy.release.worker", data["release_deploy_preview"]["template"]["key"])
|
||
self.assertEqual(11, data["release_deploy_preview"]["release_id"])
|
||
self.assertEqual(["mainland-worker-01"], data["release_deploy_preview"]["target_node_codes"])
|
||
self.assertEqual("remote-agent", data["release_deploy_preview"]["execution_mode"])
|
||
self.assertEqual(
|
||
"review_smart_rollout_preview",
|
||
data["release_deploy_preview"]["launchpad"]["launchpad_status"]["recommended_action_code"],
|
||
)
|
||
self.assertEqual("open_release_deploy_template", data["ui_intent"]["kind"])
|
||
self.assertEqual("deploy.release.worker", data["ui_intent"]["template_key"])
|
||
self.assertEqual(11, data["ui_intent"]["release_id"])
|
||
self.assertEqual(["mainland-worker-01"], data["ui_intent"]["target_node_codes"])
|
||
self.assertEqual("remote-agent", data["ui_intent"]["execution_mode"])
|
||
mock_get_release_summary.assert_called_once_with()
|
||
mock_get_release_launchpad.assert_called_once_with()
|
||
|
||
@patch("app.services.ops_service.list_managed_nodes_with_agent_state")
|
||
def test_open_release_deploy_worker_auto_switches_to_ssh_when_agent_not_ready(
|
||
self,
|
||
mock_list_managed_nodes_with_agent_state,
|
||
) -> None:
|
||
mock_list_managed_nodes_with_agent_state.return_value = {
|
||
"nodes": [
|
||
{
|
||
"node_code": "mainland-worker-01",
|
||
"is_managed": True,
|
||
"is_enabled": True,
|
||
"is_agent_online": False,
|
||
"has_ssh_access": True,
|
||
}
|
||
]
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "open_release_deploy_worker",
|
||
"node_codes": ["mainland-worker-01"],
|
||
"action_payload": {
|
||
"release_id": 11,
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("ssh", data["ui_intent"]["execution_mode"])
|
||
|
||
@patch("app.services.ops_service.get_release_launchpad")
|
||
@patch("app.services.ops_service.get_release_summary")
|
||
def test_open_rollout_dialog_carries_release_id_and_returns_preview(
|
||
self,
|
||
mock_get_release_summary,
|
||
mock_get_release_launchpad,
|
||
) -> None:
|
||
mock_get_release_summary.return_value = {
|
||
"latest_release": {
|
||
"id": 9,
|
||
"release_version": "2026.04.18",
|
||
}
|
||
}
|
||
mock_get_release_launchpad.return_value = {
|
||
"launchpad_status": {
|
||
"status": "ready",
|
||
"recommended_action_code": "publish_latest_worker",
|
||
}
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "open_rollout_dialog",
|
||
"action_payload": {
|
||
"release_id": 23,
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("rollout-dialog-preview", data["mode"])
|
||
self.assertEqual(
|
||
"publish_latest_worker",
|
||
data["rollout_dialog_preview"]["launchpad"]["launchpad_status"]["recommended_action_code"],
|
||
)
|
||
self.assertEqual(9, data["rollout_dialog_preview"]["summary"]["latest_release"]["id"])
|
||
self.assertEqual("open_rollout_dialog", data["ui_intent"]["kind"])
|
||
self.assertEqual(23, data["ui_intent"]["release_id"])
|
||
mock_get_release_summary.assert_called_once_with()
|
||
mock_get_release_launchpad.assert_called_once_with()
|
||
|
||
@patch("app.services.ops_service.create_release_and_smart_rollout_from_latest_package")
|
||
def test_create_smart_release_rollout_worker_calls_backend_flow(
|
||
self,
|
||
mock_create_release_and_smart_rollout_from_latest_package,
|
||
) -> None:
|
||
mock_create_release_and_smart_rollout_from_latest_package.return_value = (
|
||
True,
|
||
"Release rollout 已创建",
|
||
{
|
||
"release": {"id": 31, "release_version": "domaincheck_release_demo"},
|
||
"rollout_created": True,
|
||
},
|
||
)
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "create_smart_release_rollout_worker",
|
||
"action_payload": {
|
||
"channel": "stable",
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("Release rollout 已创建", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("smart-release-rollout", data["mode"])
|
||
self.assertTrue(data["smart_release_result"]["rollout_created"])
|
||
mock_create_release_and_smart_rollout_from_latest_package.assert_called_once()
|
||
called_payload = mock_create_release_and_smart_rollout_from_latest_package.call_args.args[0]
|
||
self.assertEqual("worker", called_payload["mode"])
|
||
self.assertEqual("stable", called_payload["channel"])
|
||
self.assertEqual("test/create_smart_release_rollout_worker", called_payload["created_by"])
|
||
|
||
@patch("app.services.ops_service.create_release_and_smart_rollout_from_latest_package")
|
||
def test_create_smart_release_rollout_control_passes_confirm_risky(
|
||
self,
|
||
mock_create_release_and_smart_rollout_from_latest_package,
|
||
) -> None:
|
||
mock_create_release_and_smart_rollout_from_latest_package.return_value = (
|
||
True,
|
||
"Release 已创建，智能 Rollout 需要确认",
|
||
{
|
||
"release": {"id": 32, "release_version": "domaincheck_release_demo"},
|
||
"requires_confirmation": True,
|
||
"rollout_created": False,
|
||
},
|
||
)
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "create_smart_release_rollout_control",
|
||
"action_payload": {
|
||
"confirm_risky": True,
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("Release 已创建，智能 Rollout 需要确认", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("smart-release-rollout", data["mode"])
|
||
mock_create_release_and_smart_rollout_from_latest_package.assert_called_once()
|
||
called_payload = mock_create_release_and_smart_rollout_from_latest_package.call_args.args[0]
|
||
self.assertEqual("control", called_payload["mode"])
|
||
self.assertTrue(called_payload["confirm_risky"])
|
||
|
||
@patch("app.services.ops_service.create_smart_release_rollout")
|
||
def test_create_release_rollout_worker_calls_existing_release_backend_flow(
|
||
self,
|
||
mock_create_smart_release_rollout,
|
||
) -> None:
|
||
mock_create_smart_release_rollout.return_value = (
|
||
True,
|
||
"Release rollout 已创建",
|
||
{
|
||
"release": {"id": 41, "release_version": "domaincheck_release_existing"},
|
||
"release_created": False,
|
||
"rollout_created": True,
|
||
},
|
||
)
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "create_release_rollout_worker",
|
||
"action_payload": {
|
||
"release_id": 41,
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("Release rollout 已创建", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("smart-release-rollout", data["mode"])
|
||
mock_create_smart_release_rollout.assert_called_once()
|
||
called_release_id = mock_create_smart_release_rollout.call_args.args[0]
|
||
called_payload = mock_create_smart_release_rollout.call_args.args[1]
|
||
self.assertEqual(41, called_release_id)
|
||
self.assertEqual("worker", called_payload["mode"])
|
||
self.assertEqual("test/create_release_rollout_worker", called_payload["created_by"])
|
||
|
||
@patch("app.services.ops_service.create_smart_release_rollout")
|
||
def test_create_release_rollout_control_passes_confirm_risky(
|
||
self,
|
||
mock_create_smart_release_rollout,
|
||
) -> None:
|
||
mock_create_smart_release_rollout.return_value = (
|
||
True,
|
||
"智能 Rollout 需要确认",
|
||
{
|
||
"release": {"id": 42, "release_version": "domaincheck_release_existing"},
|
||
"release_created": False,
|
||
"requires_confirmation": True,
|
||
"rollout_created": False,
|
||
},
|
||
)
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "create_release_rollout_control",
|
||
"action_payload": {
|
||
"release_id": 42,
|
||
"confirm_risky": True,
|
||
},
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("智能 Rollout 需要确认", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("smart-release-rollout", data["mode"])
|
||
mock_create_smart_release_rollout.assert_called_once()
|
||
called_release_id = mock_create_smart_release_rollout.call_args.args[0]
|
||
called_payload = mock_create_smart_release_rollout.call_args.args[1]
|
||
self.assertEqual(42, called_release_id)
|
||
self.assertEqual("control", called_payload["mode"])
|
||
self.assertTrue(called_payload["confirm_risky"])
|
||
|
||
@patch("app.services.ops_service.update_runtime_settings")
|
||
@patch("app.services.ops_service.get_runtime_settings")
|
||
def test_enable_log_sync_key_updates_runtime_settings(
|
||
self,
|
||
mock_get_runtime_settings,
|
||
mock_update_runtime_settings,
|
||
) -> None:
|
||
mock_get_runtime_settings.return_value = {
|
||
"worker_log_sync_enabled": False,
|
||
"worker_log_sync_mode": "key",
|
||
}
|
||
mock_update_runtime_settings.return_value = {
|
||
"worker_log_sync_enabled": True,
|
||
"worker_log_sync_mode": "key",
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "enable_log_sync_key",
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
mock_update_runtime_settings.assert_called_once_with(
|
||
{
|
||
"worker_log_sync_enabled": True,
|
||
"worker_log_sync_mode": "key",
|
||
}
|
||
)
|
||
self.assertEqual("远端日志回传已切到关键模式", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("settings", data["mode"])
|
||
self.assertTrue(data["runtime_settings"]["worker_log_sync_enabled"])
|
||
self.assertEqual("key", data["runtime_settings"]["worker_log_sync_mode"])
|
||
|
||
@patch("app.services.ops_service.update_runtime_settings")
|
||
@patch("app.services.ops_service.get_runtime_settings")
|
||
def test_enable_log_sync_full_updates_runtime_settings(
|
||
self,
|
||
mock_get_runtime_settings,
|
||
mock_update_runtime_settings,
|
||
) -> None:
|
||
mock_get_runtime_settings.return_value = {
|
||
"worker_log_sync_enabled": False,
|
||
"worker_log_sync_mode": "key",
|
||
}
|
||
mock_update_runtime_settings.return_value = {
|
||
"worker_log_sync_enabled": True,
|
||
"worker_log_sync_mode": "full",
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "enable_log_sync_full",
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
mock_update_runtime_settings.assert_called_once_with(
|
||
{
|
||
"worker_log_sync_enabled": True,
|
||
"worker_log_sync_mode": "full",
|
||
}
|
||
)
|
||
self.assertEqual("远端日志回传已切到全量模式", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("settings", data["mode"])
|
||
self.assertTrue(data["runtime_settings"]["worker_log_sync_enabled"])
|
||
self.assertEqual("full", data["runtime_settings"]["worker_log_sync_mode"])
|
||
|
||
@patch("app.services.ops_service.update_runtime_settings")
|
||
@patch("app.services.ops_service.get_runtime_settings")
|
||
def test_disable_log_sync_updates_runtime_settings(
|
||
self,
|
||
mock_get_runtime_settings,
|
||
mock_update_runtime_settings,
|
||
) -> None:
|
||
mock_get_runtime_settings.return_value = {
|
||
"worker_log_sync_enabled": True,
|
||
"worker_log_sync_mode": "full",
|
||
}
|
||
mock_update_runtime_settings.return_value = {
|
||
"worker_log_sync_enabled": False,
|
||
"worker_log_sync_mode": "key",
|
||
}
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "disable_log_sync",
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
mock_update_runtime_settings.assert_called_once_with(
|
||
{
|
||
"worker_log_sync_enabled": False,
|
||
"worker_log_sync_mode": "key",
|
||
}
|
||
)
|
||
self.assertEqual("远端日志回传已关闭", message)
|
||
self.assertTrue(data["handled"])
|
||
self.assertEqual("settings", data["mode"])
|
||
self.assertFalse(data["runtime_settings"]["worker_log_sync_enabled"])
|
||
self.assertEqual("key", data["runtime_settings"]["worker_log_sync_mode"])
|
||
|
||
@patch("app.services.ops_service.execute_ops_playbook")
|
||
@patch("app.services.ops_service.list_managed_nodes_with_agent_state")
|
||
def test_open_worker_logs_auto_switches_to_ssh_when_target_is_ssh_ready(
|
||
self,
|
||
mock_list_managed_nodes_with_agent_state,
|
||
mock_execute_ops_playbook,
|
||
) -> None:
|
||
mock_list_managed_nodes_with_agent_state.return_value = {
|
||
"nodes": [
|
||
{
|
||
"node_code": "mainland-worker-01",
|
||
"is_managed": True,
|
||
"is_enabled": True,
|
||
"is_agent_online": False,
|
||
"has_ssh_access": True,
|
||
}
|
||
]
|
||
}
|
||
mock_execute_ops_playbook.return_value = (True, "ok", {"run_code": "playbook-001"})
|
||
|
||
ok, message, data = execute_driver_action(
|
||
{
|
||
"action_code": "open_worker_logs_participating",
|
||
"node_codes": ["mainland-worker-01"],
|
||
"requested_by": "test",
|
||
}
|
||
)
|
||
|
||
self.assertTrue(ok, message)
|
||
self.assertEqual("ssh", data["execution_mode"])
|
||
called_payload = mock_execute_ops_playbook.call_args.args[0]
|
||
self.assertEqual("ssh", called_payload["execution_mode"])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|