import unittest from app.services.ops_release_service import ( _build_release_launchpad_status, _compact_latest_smart_rollout_preview, ) class OpsReleaseServiceLaunchpadTests(unittest.TestCase): def test_launchpad_status_blocks_when_package_missing(self) -> None: status = _build_release_launchpad_status( package={"available": False}, latest_package_preview={}, worker_preview={}, control_preview={}, ) self.assertEqual("blocked", status["status"]) self.assertEqual("release_package", status["recommended_action_code"]) self.assertEqual(status["summary"], status["summary_text"]) self.assertEqual("release_hub", status["focus_ref"]["kind"]) self.assertEqual("release_launchpad", status["focus_ref"]["section"]) def test_launchpad_status_blocks_when_rollout_preview_has_blockers(self) -> None: status = _build_release_launchpad_status( package={"available": True}, latest_package_preview={}, worker_preview={"policy_preview": {"blocked": True}}, control_preview={"policy_preview": {"blocked": False}}, ) self.assertEqual("blocked", status["status"]) self.assertEqual("fix_rollout_blockers", status["recommended_action_code"]) self.assertEqual(status["summary"], status["summary_text"]) def test_launchpad_status_blocks_when_final_signoff_missing(self) -> None: status = _build_release_launchpad_status( package={ "available": True, "final_release_ok": False, "final_release_report_available": True, "final_release_report_ok": True, "final_release_report_matches_latest": False, "final_release_report_stale_reason": "final_release_report does not match latest_release: package_name", "final_release_gate_decision": "ready", "final_release_gate_blocked_reasons": [], }, latest_package_preview={}, worker_preview={"smart_rollout": {"available": True}, "policy_preview": {}}, control_preview={"smart_rollout": {"available": True}, "policy_preview": {}}, ) self.assertEqual("blocked", status["status"]) self.assertEqual("release_prepare", status["recommended_action_code"]) self.assertIn("final_release_report does not match latest_release: package_name", status["summary"]) self.assertIn("release-prepare", status["recommended_command"]) self.assertEqual(status["summary"], status["summary_text"]) def test_launchpad_status_prefers_runtime_refresh_when_runtime_schema_is_stale(self) -> None: status = _build_release_launchpad_status( package={"available": True, "final_release_ok": True}, latest_package_preview={}, worker_preview={"smart_rollout": {"available": True}, "policy_preview": {}}, control_preview={"smart_rollout": {"available": True}, "policy_preview": {}}, runtime_build_info={ "runtime_schema_stale": True, "route_surface_complete": True, "route_surface_missing_keys": ["ops_node_handover_bootstrap_plan"], }, ) self.assertEqual("attention", status["status"]) self.assertEqual("api-restart", status["recommended_action_code"]) self.assertEqual("待刷新运行时", status["status_label"]) self.assertTrue(status["runtime_refresh_required"]) self.assertTrue(status["runtime_schema_stale"]) self.assertIn("runtime-refresh-recover", status["recommended_command"]) self.assertEqual(status["summary"], status["summary_text"]) def test_launchpad_status_marks_attention_when_confirmation_required(self) -> None: status = _build_release_launchpad_status( package={"available": True}, latest_package_preview={}, worker_preview={"requires_confirmation": True, "policy_preview": {}}, control_preview={"policy_preview": {}}, ) self.assertEqual("attention", status["status"]) self.assertEqual("review_smart_rollout_preview", status["recommended_action_code"]) self.assertEqual("release_launchpad", status["focus_ref"]["section"]) def test_launchpad_status_ready_when_worker_rollout_available(self) -> None: status = _build_release_launchpad_status( package={"available": True}, latest_package_preview={}, worker_preview={"smart_rollout": {"available": True, "execution_mode": "ssh", "execution_mode_label": "SSH"}, "policy_preview": {}}, control_preview={"smart_rollout": {"available": False}, "policy_preview": {}}, ) self.assertEqual("ready", status["status"]) self.assertEqual("publish_latest_worker", status["recommended_action_code"]) self.assertEqual("ssh", status["recommended_execution_mode"]) self.assertEqual("SSH", status["recommended_execution_mode_label"]) self.assertEqual(status["summary"], status["summary_text"]) def test_launchpad_status_requires_managed_nodes_when_no_targets_available(self) -> None: status = _build_release_launchpad_status( package={"available": True}, latest_package_preview={}, worker_preview={"smart_rollout": {"available": False}, "policy_preview": {}}, control_preview={"smart_rollout": {"available": False}, "policy_preview": {}}, ) self.assertEqual("attention", status["status"]) self.assertEqual("fix_managed_nodes", status["recommended_action_code"]) self.assertEqual("release_hub", status["focus_ref"]["kind"]) def test_launchpad_status_prefers_bootstrap_recovery_when_gap_row_has_recovery_action(self) -> None: status = _build_release_launchpad_status( package={"available": True}, latest_package_preview={}, worker_preview={ "smart_rollout": {"available": False}, "policy_preview": { "blocked": True, "release_gate": { "operational_readiness": { "rows": [ { "node_code": "mainland-worker-02", "recovery_action": "bootstrap_run", "recovery_label": "签发接入工单", "recovery_summary": "当前节点仍处于接入阶段，建议先签发 onboarding.bootstrap。", } ] } }, }, }, control_preview={"smart_rollout": {"available": False}, "policy_preview": {}}, ) self.assertEqual("blocked", status["status"]) self.assertEqual("bootstrap_run", status["recommended_action_code"]) self.assertEqual("mainland-worker-02", status["recommended_target_node_code"]) self.assertEqual("签发接入工单", status["recommended_recovery_label"]) self.assertEqual("当前节点仍处于接入阶段，建议先签发 onboarding.bootstrap。", status["recommended_recovery_summary"]) self.assertEqual(1, status["onboarding_bootstrap_pending_nodes"]) self.assertEqual(0, status["onboarding_acceptance_ready_nodes"]) self.assertEqual(1, status["onboarding_gap_nodes_total"]) self.assertIn("node-bootstrap-run", status["recommended_command"]) def test_compact_latest_smart_rollout_preview_adds_rollout_preview_alias(self) -> None: preview = _compact_latest_smart_rollout_preview( "智能 Rollout 预检通过", { "release": { "id": 18, "release_version": "2026.04.18+bd6bcb2", "channel": "stable", "status": "ready", "artifact_url": "https://example.com/domaincheck.tar.gz", }, "package": { "available": True, "package_name": "domaincheck-2026.04.18+bd6bcb2", "package_type": "tar.gz", "sha256": "sha256:test", "metadata": { "manifest": { "included_services": ["domaincheck-api", "domaincheck-worker"], } }, }, "smart_rollout": { "available": True, "execution_mode": "remote-agent", "execution_mode_label": "远端 Agent", "summary": "已按 Worker 预选 2 台节点。", "target_selector": {"node_codes": ["mainland-worker-01", "mainland-worker-02"]}, "target_nodes": [ { "node_code": "mainland-worker-01", "region": "mainland", "role": "worker", "delivery_queue_state": "healthy", }, { "node_code": "mainland-worker-02", "region": "mainland", "role": "worker", "delivery_queue_state": "retrying", }, ], }, "preview": { "target_summary": {"nodes_total": 2}, "batch_plan": {"batches_total": 2, "first_batch_size": 1}, "release_gate": {"status": "ready", "summary": "当前可推进。"}, }, }, ) self.assertEqual("remote-agent", preview["rollout_preview"]["execution_mode"]) self.assertEqual(2, preview["rollout_preview"]["target_nodes_total"]) self.assertEqual(2, preview["rollout_preview"]["expected_batches_total"]) self.assertEqual(2, preview["rollout_preview"]["expected_jobs_total"]) self.assertEqual("release_hub", preview["rollout_preview"]["focus_ref"]["kind"]) self.assertEqual("healthy", preview["smart_rollout"]["target_nodes"][0]["delivery_queue_state"]) self.assertEqual("retrying", preview["smart_rollout"]["target_nodes"][1]["delivery_queue_state"]) self.assertEqual("rollout_preview", preview["rollout_preview"]["focus_ref"]["section"]) self.assertEqual(["domaincheck-api", "domaincheck-worker"], preview["package"]["manifest"]["included_services"])