from __future__ import annotations from fastapi import FastAPI from fastapi.testclient import TestClient from src.api.v1.agents import router from src.services.delivery_closure_workbench import load_delivery_closure_workbench def test_delivery_closure_workbench_endpoint_returns_product_summary(): app = FastAPI() app.include_router(router, prefix="/api/v1") client = TestClient(app) response = client.get("/api/v1/agents/delivery-closure-workbench") assert response.status_code == 200 data = response.json() _assert_delivery_workbench_shape(data) def test_delivery_closure_workbench_uses_gitea_private_inventory_lane(): payload = load_delivery_closure_workbench() _assert_delivery_workbench_shape(payload) lanes = {lane["id"]: lane for lane in payload["lanes"]} sources = {source["id"]: source for source in payload["source_statuses"]} assert "github" not in lanes assert "github_private_backup" not in sources assert lanes["gitea_private_inventory"]["source_id"] == ( "gitea_private_inventory_p0_scorecard" ) assert lanes["gitea_private_inventory"]["blocker_count"] == 0 assert lanes["gitea_private_inventory"]["completion_percent"] == 100 assert lanes["gitea_private_inventory"]["metric"]["kind"] == "private_inventory" assert lanes["gitea_private_inventory"]["metric"]["workplan_id"] == "P0-003" assert lanes["gitea_private_inventory"]["metric"]["private_inventory_source"] == "gitea" assert lanes["gitea_private_inventory"]["metric"]["gitea_repo_inventory_status"] == "ok" assert lanes["gitea_private_inventory"]["metric"]["gitea_visibility_scope"] == "admin_export" assert lanes["gitea_private_inventory"]["metric"]["expected_product_count"] == 11 assert lanes["gitea_private_inventory"]["metric"]["present_product_row_count"] == 11 assert lanes["gitea_private_inventory"]["metric"]["missing_product_row_count"] == 0 assert ( lanes["gitea_private_inventory"]["metric"][ "authenticated_inventory_single_preflight_intake_ready" ] is True ) assert ( lanes["gitea_private_inventory"]["metric"][ "authenticated_inventory_single_preflight_intake_ready_count" ] == 1 ) assert lanes["gitea_private_inventory"]["metric"][ "authenticated_inventory_single_preflight_intake_schema_version" ] == "gitea_authenticated_inventory_single_preflight_intake_v1" assert ( lanes["gitea_private_inventory"]["metric"][ "authenticated_inventory_payload_skeleton_repo_count_floor" ] == 4 ) assert ( lanes["gitea_private_inventory"]["metric"][ "authenticated_inventory_required_redaction_attestation_count" ] == 8 ) assert ( lanes["gitea_private_inventory"]["metric"][ "authenticated_inventory_single_preflight_token_value_collection_allowed" ] is False ) assert ( lanes["gitea_private_inventory"]["metric"][ "authenticated_inventory_single_preflight_repo_write_performed" ] is False ) assert ( lanes["gitea_private_inventory"]["metric"][ "authenticated_inventory_single_preflight_refs_sync_performed" ] is False ) assert ( lanes["gitea_private_inventory"]["metric"][ "authenticated_inventory_single_preflight_github_api_used" ] is False ) assert ( lanes["gitea_private_inventory"]["metric"][ "authenticated_inventory_single_preflight_runtime_action_performed" ] is False ) assert ( lanes["gitea_private_inventory"]["metric"][ "github_lane_excluded_from_p0_blocker_count" ] is True ) assert lanes["gitea_private_inventory"]["metric"]["active_blockers"] == [] assert lanes["gitea_private_inventory"]["next_action"] == ( "continue_to_p0_006_source_to_runtime_drift_cleanup" ) def test_delivery_closure_workbench_exposes_p0_005_credential_escrow_lane(): payload = load_delivery_closure_workbench() _assert_delivery_workbench_shape(payload) lane = {lane["id"]: lane for lane in payload["lanes"]}["credential_escrow"] assert lane["source_id"] == "backup_dr_credential_escrow" assert lane["status"] == "blocked_waiting_non_secret_credential_escrow_evidence" assert lane["blocker_count"] == 5 assert lane["completion_percent"] == 0 assert lane["metric"]["kind"] == "credential_escrow_evidence" assert lane["metric"]["workplan_id"] == "P0-005" assert lane["metric"]["required_item_count"] == 5 assert lane["metric"]["effective_missing_count"] == 5 assert lane["metric"]["active_gate_present"] is True assert lane["metric"]["preflight_status"] == "blocked_waiting_owner_response_content" assert lane["metric"]["owner_response_received_count"] == 0 assert lane["metric"]["owner_response_accepted_count"] == 0 assert lane["metric"]["runtime_gate_count"] == 0 assert lane["metric"]["secret_value_collection_allowed"] is False assert lane["metric"]["credential_marker_write_authorized_count"] == 0 assert lane["metric"]["single_preflight_intake_ready"] is True assert lane["metric"]["single_preflight_intake_ready_count"] == 1 assert lane["metric"]["single_preflight_intake_schema_version"] == ( "credential_escrow_single_preflight_intake_v1" ) assert lane["metric"]["single_preflight_required_item_count"] == 5 assert ( lane["metric"]["single_preflight_secret_value_collection_allowed"] is False ) assert ( lane["metric"]["single_preflight_credential_marker_write_performed"] is False ) assert lane["metric"]["single_preflight_runtime_action_performed"] is False assert lane["metric"]["owner_response_skeleton_required_item_count"] == 5 assert ( lane["metric"]["owner_response_skeleton_secret_value_collection_allowed"] is False ) assert lane["next_action"] == ( "collect_redacted_non_secret_evidence_refs_then_rerun_preflight" ) def test_delivery_closure_workbench_exposes_p0_006_reboot_slo_lane(): payload = load_delivery_closure_workbench() _assert_delivery_workbench_shape(payload) lane = {lane["id"]: lane for lane in payload["lanes"]}["reboot_auto_recovery"] assert lane["source_id"] == "reboot_auto_recovery_slo_scorecard" assert lane["status"] == "blocked_reboot_auto_recovery_slo_not_ready" assert lane["blocker_count"] == 5 assert lane["completion_percent"] == 55 assert lane["metric"]["kind"] == "reboot_auto_recovery_slo" assert lane["metric"]["workplan_id"] == "P0-006" assert lane["metric"]["target_minutes"] == 10 assert lane["metric"]["can_claim_all_services_recovered_within_target"] is False assert lane["metric"]["observed_host_count"] == 4 assert lane["metric"]["missing_host_count"] == 0 assert lane["metric"]["unreachable_host_count"] == 0 assert lane["metric"]["stale_host_count"] == 4 assert lane["metric"]["service_green"] is False assert lane["metric"]["product_data_green"] is False assert lane["metric"]["backup_core_green"] is True assert lane["metric"]["stockplatform_freshness_status"] == "blocked" assert lane["metric"]["stockplatform_ingestion_status"] == "ok" assert lane["metric"]["stockplatform_freshness_blocker_count"] == 1 assert lane["metric"]["stockplatform_ingestion_blocker_count"] == 0 assert lane["metric"]["stockplatform_final_retry_window_passed"] is False assert lane["metric"]["stockplatform_controlled_recovery_gate_required"] is False assert lane["metric"]["host_reboot_performed"] is False assert lane["metric"]["service_restart_performed"] is False assert lane["metric"]["database_write_or_restore_performed"] is False assert lane["metric"]["secret_value_collection_allowed"] is False assert "stockplatform_freshness_blocked" in lane["metric"]["active_blockers"] assert lane["next_action"] == ( "wait_ai_recommendations_refresh_then_rerun_slo_verify_only_" "no_reboot_no_db_write" ) def _assert_delivery_workbench_shape(data: dict): assert data["schema_version"] == "delivery_closure_workbench_v1" assert data["summary"]["source_count"] == 8 assert data["summary"]["loaded_source_count"] == 8 assert data["summary"]["runtime_execution_authorized"] is False assert data["summary"]["remote_write_authorized"] is False assert data["summary"]["repo_creation_authorized"] is False assert data["summary"]["visibility_change_authorized"] is False assert data["summary"]["refs_sync_authorized"] is False assert data["summary"]["workflow_trigger_authorized"] is False assert data["summary"]["github_global_freeze_enabled"] is True assert data["summary"]["github_lane_status"] == "stopped_retired_do_not_use" assert data["summary"]["github_lane_excluded_from_p0_blocker_count"] is True assert data["summary"]["github_blocked_preflight_target_count"] == 0 assert data["summary"]["github_operator_unblock_required"] is False assert data["summary"]["reboot_auto_recovery_status"] == ( "blocked_reboot_auto_recovery_slo_not_ready" ) assert data["summary"]["reboot_auto_recovery_workplan_id"] == "P0-006" assert data["summary"]["reboot_auto_recovery_readiness_percent"] == 55 assert data["summary"]["reboot_auto_recovery_active_blocker_count"] == 5 assert data["summary"]["reboot_auto_recovery_can_claim_slo"] is False assert data["summary"]["reboot_auto_recovery_service_green"] is False assert data["summary"]["reboot_auto_recovery_product_data_green"] is False assert data["summary"]["reboot_auto_recovery_observed_host_count"] == 4 assert data["summary"]["reboot_auto_recovery_stale_host_count"] == 4 assert ( data["summary"]["reboot_auto_recovery_stockplatform_freshness_status"] == "blocked" ) assert ( data["summary"]["reboot_auto_recovery_stockplatform_ingestion_status"] == "ok" ) assert ( data["summary"][ "reboot_auto_recovery_stockplatform_final_retry_window_passed" ] is False ) assert ( data["summary"][ "reboot_auto_recovery_stockplatform_controlled_recovery_gate_required" ] is False ) assert data["summary"]["reboot_auto_recovery_safe_next_step"] == ( "wait_ai_recommendations_refresh_then_rerun_slo_verify_only_" "no_reboot_no_db_write" ) assert data["summary"]["gitea_private_inventory_status"] == ( "closed_gitea_private_inventory_controlled_closeout" ) assert data["summary"]["gitea_private_inventory_workplan_id"] == "P0-003" assert data["summary"]["gitea_private_inventory_source"] == "gitea" assert data["summary"]["gitea_private_inventory_review_readiness_percent"] == 100 assert data["summary"]["gitea_private_inventory_active_blocker_count"] == 0 assert data["summary"]["gitea_private_inventory_repo_inventory_status"] == "ok" assert data["summary"]["gitea_private_inventory_visibility_scope"] == "admin_export" assert data["summary"]["gitea_private_inventory_expected_product_count"] == 11 assert data["summary"]["gitea_private_inventory_present_product_row_count"] == 11 assert data["summary"]["gitea_private_inventory_missing_product_row_count"] == 0 assert data["summary"]["gitea_private_inventory_accepted_payload_count"] == 1 assert ( data["summary"][ "gitea_private_inventory_owner_coverage_attestation_received_count" ] == 1 ) assert ( data["summary"][ "gitea_private_inventory_authenticated_single_preflight_intake_ready" ] is True ) assert ( data["summary"][ "gitea_private_inventory_authenticated_single_preflight_intake_ready_count" ] == 1 ) assert data["summary"][ "gitea_private_inventory_authenticated_single_preflight_intake_schema_version" ] == "gitea_authenticated_inventory_single_preflight_intake_v1" assert ( data["summary"][ "gitea_private_inventory_authenticated_payload_skeleton_repo_count_floor" ] == 4 ) assert ( data["summary"][ "gitea_private_inventory_authenticated_required_redaction_attestation_count" ] == 8 ) assert ( data["summary"][ "gitea_private_inventory_authenticated_single_preflight_token_value_collection_allowed" ] is False ) assert ( data["summary"][ "gitea_private_inventory_authenticated_single_preflight_repo_write_performed" ] is False ) assert ( data["summary"][ "gitea_private_inventory_authenticated_single_preflight_refs_sync_performed" ] is False ) assert ( data["summary"][ "gitea_private_inventory_authenticated_single_preflight_github_api_used" ] is False ) assert ( data["summary"][ "gitea_private_inventory_authenticated_single_preflight_runtime_action_performed" ] is False ) assert ( data["summary"][ "gitea_private_inventory_all_active_product_repos_have_owner_readiness_row" ] is True ) assert data["summary"]["p0_cicd_baseline_status"] == ( "ready_for_template_copy_apply_gate" ) assert data["summary"]["p0_cicd_baseline_source_readiness_percent"] == 100 assert data["summary"]["production_deploy_status"] == "closure_verified" assert data["summary"]["production_deploy_image_tag_matches_main"] is True assert data["summary"]["backup_credential_escrow_intake_status"] == ( "blocked_waiting_non_secret_credential_escrow_evidence" ) assert data["summary"]["backup_credential_escrow_required_item_count"] == 5 assert data["summary"]["backup_credential_escrow_effective_missing_count"] == 5 assert data["summary"]["backup_credential_escrow_secret_value_collection_allowed"] is False assert data["summary"]["backup_credential_marker_write_authorized_count"] == 0 assert ( data["summary"][ "backup_credential_escrow_single_preflight_intake_ready" ] is True ) assert ( data["summary"][ "backup_credential_escrow_single_preflight_intake_ready_count" ] == 1 ) assert data["summary"][ "backup_credential_escrow_single_preflight_intake_schema_version" ] == "credential_escrow_single_preflight_intake_v1" assert ( data["summary"][ "backup_credential_escrow_single_preflight_required_item_count" ] == 5 ) assert data["summary"][ "backup_credential_escrow_single_preflight_safe_next_step" ] == "collect_redacted_non_secret_evidence_refs_then_rerun_preflight" assert ( data["summary"][ "backup_credential_escrow_single_preflight_secret_value_collection_allowed" ] is False ) assert ( data["summary"][ "backup_credential_escrow_single_preflight_credential_marker_write_performed" ] is False ) assert ( data["summary"][ "backup_credential_escrow_single_preflight_runtime_action_performed" ] is False ) assert data["summary"]["secret_values_collected"] is False lane_ids = {lane["id"] for lane in data["lanes"]} assert lane_ids == { "release", "production_deploy", "reboot_auto_recovery", "credential_escrow", "gitea_private_inventory", "cicd_baseline", "gitea", "runtime", "backup", } assert data["operation_boundaries"]["read_only_api_allowed"] is True assert data["operation_boundaries"]["runtime_write_allowed"] is False assert data["operation_boundaries"]["remote_write_allowed"] is False assert data["operation_boundaries"]["repo_creation_allowed"] is False assert data["operation_boundaries"]["visibility_change_allowed"] is False assert data["operation_boundaries"]["refs_sync_allowed"] is False assert data["operation_boundaries"]["workflow_trigger_allowed"] is False assert data["operation_boundaries"]["gitea_api_write_allowed"] is False assert ( data["operation_boundaries"][ "gitea_authenticated_inventory_import_execution_allowed" ] is False ) assert data["operation_boundaries"]["github_write_channel_ready"] is False assert data["operation_boundaries"]["github_controlled_apply_allowed"] is False assert data["operation_boundaries"]["secret_value_collection_allowed"] is False assert data["operation_boundaries"]["host_reboot_performed"] is False assert data["operation_boundaries"]["service_restart_performed"] is False assert data["operation_boundaries"]["database_write_or_restore_performed"] is False assert ( data["operation_boundaries"]["stockplatform_manual_data_write_performed"] is False )