Files
awoooi/apps/api/tests/test_delivery_closure_workbench_api.py
Your Name 005be30fb1
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 19s
CD Pipeline / build-and-deploy (push) Successful in 5m34s
CD Pipeline / post-deploy-checks (push) Successful in 1m12s
feat(api): close credential escrow evidence readback
2026-06-29 21:26:50 +08:00

455 lines
18 KiB
Python

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"] == "credential_escrow_evidence_intake_readiness"
assert lane["status"] == "closed_credential_escrow_evidence_refs_controlled_closeout"
assert lane["blocker_count"] == 0
assert lane["completion_percent"] == 100
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"] == 0
assert lane["metric"]["active_gate_present"] is False
assert lane["metric"]["preflight_status"] == "ready_for_reviewer_acceptance_writeback"
assert lane["metric"]["accepted_item_count"] == 5
assert lane["metric"]["owner_response_received_count"] == 1
assert lane["metric"]["owner_response_accepted_count"] == 1
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"]["controlled_closeout_status"] == (
"ready_for_p0_005_controlled_closeout"
)
assert (
lane["metric"]["controlled_closeout_redacted_receipt_writeback_ready_count"]
== 1
)
assert lane["metric"]["controlled_closeout_source_ref"] == (
"docs/operations/"
"awoooi-credential-escrow-evidence-controlled-closeout-receipt.snapshot.json"
)
assert lane["metric"]["controlled_closeout_projected_effective_missing_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"] == "continue_to_p0_006_source_to_runtime_drift_cleanup"
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"] == 1
assert lane["completion_percent"] == 82
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 True
assert lane["metric"]["product_data_green"] is True
assert lane["metric"]["backup_core_green"] is True
assert lane["metric"]["stockplatform_freshness_status"] == "ok"
assert lane["metric"]["stockplatform_ingestion_status"] == "ok"
assert lane["metric"]["stockplatform_freshness_blocker_count"] == 0
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 lane["metric"]["active_blockers"] == [
"host_boot_observation_older_than_target_window"
]
assert lane["next_action"] == (
"timer_and_service_data_readback_green_wait_for_next_all_host_reboot_event_"
"or_approved_reboot_drill_to_prove_10_minute_slo"
)
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"] == 82
assert data["summary"]["reboot_auto_recovery_active_blocker_count"] == 1
assert data["summary"]["reboot_auto_recovery_can_claim_slo"] is False
assert data["summary"]["reboot_auto_recovery_service_green"] is True
assert data["summary"]["reboot_auto_recovery_product_data_green"] is True
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"]
== "ok"
)
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"] == (
"timer_and_service_data_readback_green_wait_for_next_all_host_reboot_event_"
"or_approved_reboot_drill_to_prove_10_minute_slo"
)
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"] == (
"closed_credential_escrow_evidence_refs_controlled_closeout"
)
assert data["summary"]["backup_credential_escrow_active_gate_present"] is False
assert data["summary"]["backup_credential_escrow_preflight_status"] == (
"ready_for_reviewer_acceptance_writeback"
)
assert data["summary"]["backup_credential_escrow_required_item_count"] == 5
assert data["summary"]["backup_credential_escrow_effective_missing_count"] == 0
assert data["summary"]["backup_credential_escrow_accepted_item_count"] == 5
assert data["summary"]["backup_credential_escrow_owner_response_received_count"] == 1
assert data["summary"]["backup_credential_escrow_owner_response_accepted_count"] == 1
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_controlled_closeout_status"] == (
"ready_for_p0_005_controlled_closeout"
)
assert (
data["summary"][
"backup_credential_escrow_redacted_receipt_writeback_ready_count"
]
== 1
)
assert data["summary"]["backup_credential_escrow_closeout_receipt_ref"] == (
"docs/operations/"
"awoooi-credential-escrow-evidence-controlled-closeout-receipt.snapshot.json"
)
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"
] == "continue_to_p0_006_source_to_runtime_drift_cleanup"
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
)