diff --git a/apps/api/src/services/gitea_capability_ci_cd_optimization_readback.py b/apps/api/src/services/gitea_capability_ci_cd_optimization_readback.py index 3931a9b81..7515bfb46 100644 --- a/apps/api/src/services/gitea_capability_ci_cd_optimization_readback.py +++ b/apps/api/src/services/gitea_capability_ci_cd_optimization_readback.py @@ -325,6 +325,10 @@ def build_gitea_capability_ci_cd_optimization_readback( alert_chain = _dict(gitea_cicd_alert_receipt_chain_readback) alert_chain_ready = alert_chain.get("ready") is True alert_chain_summary = _dict(alert_chain.get("summary")) + webhook_hmac_source_ready = ( + alert_chain_ready + or alert_chain_summary.get("gitea_webhook_hmac_source_ready") is True + ) professional_feature_rollups = _dict( workflow_source_inventory.get("professional_feature_rollups") ) @@ -352,6 +356,19 @@ def build_gitea_capability_ci_cd_optimization_readback( source_ready = _int(cicd_rollups.get("missing_required_source_count")) == 0 quiet_notification_count = _int( notification_rollups.get("notification_contracts_quiet_success_count") + ) or len(_list(notification_rollups.get("suppressed_success_rule_ids"))) + dora_delivery_readback = _build_dora_delivery_performance_readback( + dev_prod_summary=dev_prod_summary, + workflow_source_count=workflow_source_count, + workflow_contract_count=workflow_contract_count, + source_ready=source_ready, + known_workflows=known_workflows, + professional_feature_rollups=professional_feature_rollups, + alert_chain_ready=alert_chain_ready, + alert_chain_summary=alert_chain_summary, + ) + dora_source_contract_ready = ( + dora_delivery_readback.get("source_contract_ready") is True ) capabilities = [ @@ -542,10 +559,30 @@ def build_gitea_capability_ci_cd_optimization_readback( "webhooks_integrations", "integration", "Webhooks and notification integrations", - "partially_ready" if quiet_notification_count > 0 else "action_required", - "notification policy exists; delivery receipt chain still needs live closure", - ["GET /api/v1/agents/backup-notification-policy"], - "attach webhook delivery and Telegram/AwoooP receipt readback to backup/CD events", + "source_ready" + if quiet_notification_count > 0 and webhook_hmac_source_ready + else "partially_ready" + if quiet_notification_count > 0 + else "action_required", + "notification policy and Gitea webhook HMAC source contract exist; live delivery receipts remain separate" + if quiet_notification_count > 0 and webhook_hmac_source_ready + else "notification policy exists; delivery receipt chain still needs live closure", + [ + "GET /api/v1/agents/backup-notification-policy", + "GET /api/v1/agents/gitea-cicd-alert-receipt-chain-readback", + ], + "bind live Gitea delivery IDs and receipt stages without reading webhook secret values" + if webhook_hmac_source_ready + else "attach webhook delivery and Telegram/AwoooP receipt readback to backup/CD events", + { + "quiet_notification_policy_count": quiet_notification_count, + "gitea_webhook_hmac_source_ready": webhook_hmac_source_ready, + "gitea_webhook_delivery_header_supported": alert_chain_summary.get( + "gitea_webhook_delivery_header_supported" + ) + is True, + "live_delivery_receipt_bound": False, + }, ), _capability( "branch_protection_required_checks", @@ -893,21 +930,27 @@ def build_gitea_capability_ci_cd_optimization_readback( "gitea_125_webhook_hmac_delivery", "Gitea 1.25 Repository", "source_ready" - if quiet_notification_count > 0 and alert_chain_ready + if quiet_notification_count > 0 and webhook_hmac_source_ready else "action_required", "Webhook HMAC delivery and notification receipt chain", - "notification policy and alert receipt chain are ready" - if quiet_notification_count > 0 and alert_chain_ready + "Gitea webhook HMAC source contract and delivery header readback are wired; live delivery receipts remain receipt work" + if quiet_notification_count > 0 and webhook_hmac_source_ready else "webhook delivery receipts need live closure", [ "GET /api/v1/agents/gitea-cicd-alert-receipt-chain-readback", "GET /api/v1/agents/backup-notification-policy", "https://docs.gitea.com/1.25/usage/repository/webhooks", ], - "bind Gitea webhook delivery ID, signature verification result, and Telegram/AwoooP receipt stage into one readback", + "bind live Gitea delivery ID, signature verification result, and Telegram/AwoooP receipt stage into one readback", { "quiet_notification_policy_count": quiet_notification_count, "alert_receipt_chain_ready": alert_chain_ready, + "gitea_webhook_hmac_source_ready": webhook_hmac_source_ready, + "gitea_webhook_delivery_header_supported": alert_chain_summary.get( + "gitea_webhook_delivery_header_supported" + ) + is True, + "live_delivery_receipt_bound": False, }, ), _professional_practice( @@ -984,15 +1027,34 @@ def build_gitea_capability_ci_cd_optimization_readback( _professional_practice( "dora_delivery_metrics_receipts", "DORA", - "action_required", + "source_ready" if dora_source_contract_ready else "action_required", "Five-metric delivery performance readback", - "deploy markers and alert receipts exist, but DORA trend metrics are not yet computed per product", + "DORA five-metric source contract covers product repositories; trend values remain receipt-backed work" + if dora_source_contract_ready + else "deploy markers and alert receipts exist, but DORA trend metrics are not yet computed per product", [ "GET /api/v1/agents/p0-cicd-baseline-source-readiness", "GET /api/v1/agents/gitea-cicd-alert-receipt-chain-readback", "https://dora.dev/guides/dora-metrics/", ], "derive change lead time, deployment frequency, failed deployment recovery time, change fail rate, and deployment rework rate from Gitea/CD receipts", + { + "metric_count": _int(dora_delivery_readback.get("metric_count")), + "source_ready_metric_count": _int( + dora_delivery_readback.get("source_ready_metric_count") + ), + "per_product_contract_count": _int( + dora_delivery_readback.get("per_product_contract_count") + ), + "trend_values_computed": dora_delivery_readback.get( + "trend_values_computed" + ) + is True, + "receipt_backed_computation_required": dora_delivery_readback.get( + "receipt_backed_computation_required" + ) + is True, + }, ), _professional_practice( "slsa_build_provenance_isolation", @@ -1195,6 +1257,7 @@ def build_gitea_capability_ci_cd_optimization_readback( }, "rollups": rollups, "workflow_source_inventory": workflow_source_inventory, + "delivery_performance_readback": dora_delivery_readback, "capability_rows": capabilities, "ci_cd_optimization_rows": optimizations, "professional_ci_cd_practice_rows": professional_practices, @@ -1203,6 +1266,110 @@ def build_gitea_capability_ci_cd_optimization_readback( return payload +def _build_dora_delivery_performance_readback( + *, + dev_prod_summary: dict[str, Any], + workflow_source_count: int, + workflow_contract_count: int, + source_ready: bool, + known_workflows: dict[str, Any], + professional_feature_rollups: dict[str, Any], + alert_chain_ready: bool, + alert_chain_summary: dict[str, Any], +) -> dict[str, Any]: + metric_specs = [ + ( + "change_lead_time", + "Change lead time", + "commit timestamp to deployed marker timestamp", + "derive commit-to-production interval from Gitea commit, CD run, deploy marker, and production readback receipts", + ), + ( + "deployment_frequency", + "Deployment frequency", + "successful deploy markers per product and environment", + "aggregate deploy marker cadence per product, dev branch, and main branch without counting canceled runs as deploys", + ), + ( + "failed_deployment_recovery_time", + "Failed deployment recovery time", + "failed CD or failed production readback to next verified deploy", + "bind failed Gitea run/readback receipts to the next successful deploy marker and smoke receipt", + ), + ( + "change_fail_rate", + "Change fail rate", + "failed deployment receipts divided by deployment attempts", + "classify Gitea CD failures, canceled superseded runs, and production readback failures separately before computing rate", + ), + ( + "deployment_rework_rate", + "Deployment rework rate", + "follow-up fix/revert/redeploy markers per delivered change", + "derive rework from commit subjects, deploy markers, and post-deploy alert receipts rather than manual labels", + ), + ] + per_product_contract_count = _int(dev_prod_summary.get("expected_product_count")) + workflow_contract_fresh = ( + workflow_source_count > 0 and workflow_source_count == workflow_contract_count + ) + deploy_marker_contract_ready = ( + known_workflows.get("has_cd_pipeline") is True + and _int(professional_feature_rollups.get("argocd_gitops_workflow_count")) > 0 + ) + notification_contract_ready = ( + alert_chain_ready + or _int(alert_chain_summary.get("workflow_notify_bridge_count")) > 0 + or alert_chain_summary.get("gitea_webhook_hmac_source_ready") is True + ) + source_contract_ready = ( + source_ready + and per_product_contract_count > 0 + and workflow_contract_fresh + and deploy_marker_contract_ready + and notification_contract_ready + ) + metric_rows = [ + { + "metric_id": metric_id, + "title": title, + "status": "source_ready" if source_contract_ready else "action_required", + "measurement_contract": measurement_contract, + "next_action": next_action, + "evidence_refs": [ + ".gitea/workflows/cd.yaml", + "GET /api/v1/agents/p0-cicd-baseline-source-readiness", + "GET /api/v1/agents/gitea-cicd-alert-receipt-chain-readback", + "https://dora.dev/guides/dora-metrics/", + ], + "trend_value_available": False, + } + for metric_id, title, measurement_contract, next_action in metric_specs + ] + return { + "framework": "DORA", + "source_contract_ready": source_contract_ready, + "metric_count": len(metric_rows), + "source_ready_metric_count": sum( + 1 for row in metric_rows if row["status"] == "source_ready" + ), + "per_product_contract_count": per_product_contract_count, + "workflow_contract_fresh": workflow_contract_fresh, + "deploy_marker_contract_ready": deploy_marker_contract_ready, + "notification_contract_ready": notification_contract_ready, + "trend_values_computed": False, + "receipt_backed_computation_required": True, + "operation_boundaries": { + "read_only_api_allowed": True, + "workflow_trigger_allowed": False, + "gitea_live_write_allowed": False, + "secret_value_read_allowed": False, + "github_api_allowed": False, + }, + "metric_rows": metric_rows, + } + + def _capability( capability_id: str, domain: str, @@ -1448,6 +1615,26 @@ def _require_payload(payload: dict[str, Any], label: str) -> None: "status", ): raise ValueError(f"{label}: professional_ci_cd_practice_status_counts mismatch") + dora_readback = _dict(payload.get("delivery_performance_readback")) + dora_metric_rows = _list(dora_readback.get("metric_rows")) + dora_metric_ids = { + str(row.get("metric_id") or "") + for row in dora_metric_rows + if isinstance(row, dict) + } + expected_dora_metric_ids = { + "change_lead_time", + "deployment_frequency", + "failed_deployment_recovery_time", + "change_fail_rate", + "deployment_rework_rate", + } + if dora_metric_ids != expected_dora_metric_ids: + raise ValueError(f"{label}: DORA metric rows incomplete") + if dora_readback.get("metric_count") != len(dora_metric_rows): + raise ValueError(f"{label}: DORA metric_count mismatch") + if dora_readback.get("trend_values_computed") is not False: + raise ValueError(f"{label}: DORA trend values must remain explicit") _require_no_plaintext_secret_keys(payload, label) diff --git a/apps/api/src/services/gitea_cicd_alert_receipt_chain_readback.py b/apps/api/src/services/gitea_cicd_alert_receipt_chain_readback.py index e6ca4f7c1..aad972531 100644 --- a/apps/api/src/services/gitea_cicd_alert_receipt_chain_readback.py +++ b/apps/api/src/services/gitea_cicd_alert_receipt_chain_readback.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio from datetime import datetime +from pathlib import Path from typing import Any from zoneinfo import ZoneInfo @@ -16,12 +17,14 @@ from src.services.gitea_full_server_backup_dr_receipts import ( from src.services.gitea_workflow_runner_health import ( load_latest_gitea_workflow_runner_health, ) +from src.services.snapshot_paths import resolve_repo_root from src.services.telegram_alert_monitoring_coverage_readback import ( load_latest_telegram_alert_monitoring_coverage_readback, ) SCHEMA_VERSION = "gitea_cicd_alert_receipt_chain_readback_v1" TZ_TAIPEI = ZoneInfo("Asia/Taipei") +_DEFAULT_REPO_ROOT = resolve_repo_root(Path(__file__)) async def load_latest_gitea_cicd_alert_receipt_chain_readback( @@ -49,6 +52,7 @@ def build_gitea_cicd_alert_receipt_chain_readback( backup_notification_policy: dict[str, Any], workflow_runner_health: dict[str, Any], full_server_backup_dr_receipts: dict[str, Any], + repo_root: Path | None = None, generated_at: str | None = None, ) -> dict[str, Any]: """Compose CI/CD and backup alert receipt chain coverage.""" @@ -58,7 +62,9 @@ def build_gitea_cicd_alert_receipt_chain_readback( workflow_rollups = _dict(workflow_runner_health.get("rollups")) full_dr_summary = _dict(full_server_backup_dr_receipts.get("summary")) - monitoring_gap_count = _int(coverage_summary.get("monitoring_live_receipt_gap_count")) + monitoring_gap_count = _int( + coverage_summary.get("monitoring_live_receipt_gap_count") + ) full_dr_gap_count = _int(full_server_backup_dr_receipts.get("active_gap_count")) full_dr_gap_ids = _list(full_server_backup_dr_receipts.get("active_gap_ids")) full_dr_restore_drill_only = ( @@ -76,18 +82,21 @@ def build_gitea_cicd_alert_receipt_chain_readback( and _int(coverage_summary.get("alert_receipt_pipeline_ready_count")) > 0 ) ai_context_verified_ready = ( - ( - coverage_summary.get("ai_loop_context_ready") is True - and _int(coverage_summary.get("ai_loop_ai_agent_context_receipt_count")) > 0 - ) - or _int(coverage_summary.get("verified_ai_agent_context_receipt_count")) > 0 + coverage_summary.get("ai_loop_context_ready") is True + and _int(coverage_summary.get("ai_loop_ai_agent_context_receipt_count")) > 0 + ) or _int(coverage_summary.get("verified_ai_agent_context_receipt_count")) > 0 + webhook_hmac_source = _build_gitea_webhook_hmac_source_readiness( + repo_root or _DEFAULT_REPO_ROOT ) + webhook_hmac_source_ready = webhook_hmac_source.get("source_ready") is True receipt_rows = [ { "chain_id": "gitea_cd_pipeline_notifications", "domain": "cicd", - "status": "source_ready" if notify_workflow_count > 0 else "action_required", + "status": "source_ready" + if notify_workflow_count > 0 + else "action_required", "receipt_state": "notify_bridge_source_present", "evidence_refs": [ "GET /api/v1/agents/gitea-workflow-runner-health", @@ -95,6 +104,26 @@ def build_gitea_cicd_alert_receipt_chain_readback( ], "next_action": "bind CD notify bridge delivery receipts to deployed marker readback", }, + { + "chain_id": "gitea_webhook_hmac_delivery", + "domain": "gitea", + "status": "source_ready" + if webhook_hmac_source_ready + else "action_required", + "receipt_state": "hmac_delivery_source_contract_ready" + if webhook_hmac_source_ready + else "hmac_delivery_source_contract_incomplete", + "evidence_refs": [ + "apps/api/src/api/v1/gitea_webhook.py", + "apps/api/tests/test_gitea_webhook.py", + "https://docs.gitea.com/1.25/usage/repository/webhooks", + ], + "next_action": ( + "bind live Gitea delivery IDs, signature decisions, and " + "notification receipt stages into a single metadata-only receipt" + ), + "metrics": webhook_hmac_source, + }, { "chain_id": "backup_notification_policy", "domain": "backup", @@ -122,7 +151,9 @@ def build_gitea_cicd_alert_receipt_chain_readback( else "source_contract_ready_restore_drill_pending" if full_dr_restore_drill_only else "component_receipts_pending", - "evidence_refs": ["GET /api/v1/agents/gitea-full-server-backup-dr-receipts"], + "evidence_refs": [ + "GET /api/v1/agents/gitea-full-server-backup-dr-receipts" + ], "next_action": "run a controlled restore-drill verifier before DR complete" if full_dr_restore_drill_only else "close full-server component receipt rows before DR complete", @@ -200,9 +231,7 @@ def build_gitea_cicd_alert_receipt_chain_readback( ] active_gaps = [ - row["chain_id"] - for row in receipt_rows - if row["status"] == "action_required" + row["chain_id"] for row in receipt_rows if row["status"] == "action_required" ] payload = { "schema_version": SCHEMA_VERSION, @@ -256,6 +285,15 @@ def build_gitea_cicd_alert_receipt_chain_readback( "workflow_notify_bridge_count": notify_workflow_count, "backup_quiet_success_contract_count": quiet_contract_count, "backup_immediate_escalation_rule_count": immediate_escalation_count, + "gitea_webhook_hmac_source_ready": webhook_hmac_source_ready, + "gitea_webhook_delivery_header_supported": webhook_hmac_source.get( + "delivery_header_captured" + ) + is True, + "gitea_webhook_signature_verification_source_ready": webhook_hmac_source.get( + "signature_verification_source_ready" + ) + is True, }, "receipt_chain_rows": receipt_rows, "safe_next_actions": [ @@ -268,6 +306,68 @@ def build_gitea_cicd_alert_receipt_chain_readback( return payload +def _build_gitea_webhook_hmac_source_readiness(repo_root: Path) -> dict[str, Any]: + route_path = repo_root / "apps/api/src/api/v1/gitea_webhook.py" + test_path = repo_root / "apps/api/tests/test_gitea_webhook.py" + route_text = _read_text(route_path) + test_text = _read_text(test_path) + + endpoint_source_present = bool(route_text) + test_source_present = bool(test_text) + signature_header_required = ( + "X-Gitea-Signature" in route_text + and "Missing X-Gitea-Signature" in route_text + and "Invalid signature" in route_text + ) + delivery_header_captured = ( + "X-Gitea-Delivery" in route_text and "delivery_id" in route_text + ) + constant_time_compare = "hmac.compare_digest" in route_text + whitelist_guard_present = "verify_gitea_allowed_repo" in route_text + supported_event_count = sum( + 1 + for marker in ( + 'x_gitea_event == "pull_request"', + 'x_gitea_event == "push"', + 'x_gitea_event == "workflow_run"', + 'x_gitea_event == "ping"', + ) + if marker in route_text + ) + verified_by_test = ( + "test_webhook_valid_signature" in test_text + and "test_webhook_invalid_signature" in test_text + and "test_webhook_missing_signature" in test_text + ) + signature_verification_source_ready = ( + endpoint_source_present + and signature_header_required + and constant_time_compare + and verified_by_test + ) + source_ready = ( + signature_verification_source_ready + and delivery_header_captured + and whitelist_guard_present + and supported_event_count >= 4 + ) + return { + "source_ready": source_ready, + "endpoint_source_present": endpoint_source_present, + "test_source_present": test_source_present, + "signature_header_required": signature_header_required, + "signature_verification_source_ready": signature_verification_source_ready, + "delivery_header_captured": delivery_header_captured, + "constant_time_compare": constant_time_compare, + "repo_whitelist_guard_present": whitelist_guard_present, + "supported_event_count": supported_event_count, + "valid_invalid_missing_signature_tests_present": verified_by_test, + "secret_value_read_allowed": False, + "raw_payload_storage_allowed": False, + "live_delivery_receipt_bound": False, + } + + def _require_payload(payload: dict[str, Any], label: str) -> None: if payload.get("schema_version") != SCHEMA_VERSION: raise ValueError(f"{label}: invalid schema_version") @@ -287,9 +387,13 @@ def _require_payload(payload: dict[str, Any], label: str) -> None: "github_api_allowed", "host_or_k8s_write_allowed", } - unexpected = sorted(flag for flag in blocked_flags if boundaries.get(flag) is not False) + unexpected = sorted( + flag for flag in blocked_flags if boundaries.get(flag) is not False + ) if unexpected: - raise ValueError(f"{label}: blocked operation flags must be false: {unexpected}") + raise ValueError( + f"{label}: blocked operation flags must be false: {unexpected}" + ) rows = _list(payload.get("receipt_chain_rows")) summary = _dict(payload.get("summary")) if summary.get("receipt_chain_count") != len(rows): @@ -311,3 +415,10 @@ def _int(value: Any) -> int: return int(value) except (TypeError, ValueError): return 0 + + +def _read_text(path: Path) -> str: + try: + return path.read_text(encoding="utf-8") + except OSError: + return "" diff --git a/apps/api/tests/test_gitea_capability_ci_cd_optimization_readback_api.py b/apps/api/tests/test_gitea_capability_ci_cd_optimization_readback_api.py index 8f08ef5d0..bef907945 100644 --- a/apps/api/tests/test_gitea_capability_ci_cd_optimization_readback_api.py +++ b/apps/api/tests/test_gitea_capability_ci_cd_optimization_readback_api.py @@ -195,6 +195,19 @@ def test_gitea_capability_readback_rolls_up_complete_feature_surface() -> None: capability_by_id["branch_protection_required_checks"]["status"] == "source_ready" ) + assert capability_by_id["webhooks_integrations"]["status"] == "source_ready" + assert ( + capability_by_id["webhooks_integrations"]["metrics"][ + "gitea_webhook_hmac_source_ready" + ] + is True + ) + assert ( + capability_by_id["webhooks_integrations"]["metrics"][ + "live_delivery_receipt_bound" + ] + is False + ) assert capability_by_id["backup_dr_restore"]["status"] == "source_ready" assert ( capability_by_id["backup_dr_restore"]["metrics"][ @@ -254,12 +267,10 @@ def test_gitea_capability_readback_prioritizes_professional_cicd_gaps() -> None: payload["professional_ci_cd_practice_rows"] ) assert payload["rollups"]["p0_professional_ci_cd_practice_gap_count"] == 0 - assert payload["rollups"]["market_professional_ci_cd_gap_count"] == 5 + assert payload["rollups"]["market_professional_ci_cd_gap_count"] == 3 assert payload["rollups"]["all_gitea_features_fully_utilized"] is False assert payload["rollups"]["professional_ci_cd_practice_gap_ids"] == [ - "gitea_125_webhook_hmac_delivery", "gitea_125_packages_release_registry", - "dora_delivery_metrics_receipts", "slsa_build_provenance_isolation", "artifact_cache_sbom_retention", ] @@ -278,13 +289,45 @@ def test_gitea_capability_readback_prioritizes_professional_cicd_gaps() -> None: practice_by_id["gitea_125_actions_compatibility_gap_guard"]["status"] == "source_ready" ) + assert practice_by_id["gitea_125_webhook_hmac_delivery"]["status"] == "source_ready" + assert ( + practice_by_id["gitea_125_webhook_hmac_delivery"]["metrics"][ + "gitea_webhook_hmac_source_ready" + ] + is True + ) + assert ( + practice_by_id["gitea_125_webhook_hmac_delivery"]["metrics"][ + "live_delivery_receipt_bound" + ] + is False + ) assert ( practice_by_id["gitops_argocd_automated_sync_contract"]["status"] == "source_ready" ) assert practice_by_id["otel_cicd_pipeline_telemetry"]["status"] == "source_ready" + assert practice_by_id["dora_delivery_metrics_receipts"]["status"] == "source_ready" assert ( - practice_by_id["dora_delivery_metrics_receipts"]["status"] == "action_required" + practice_by_id["dora_delivery_metrics_receipts"]["metrics"]["metric_count"] == 5 + ) + assert ( + practice_by_id["dora_delivery_metrics_receipts"]["metrics"][ + "source_ready_metric_count" + ] + == 5 + ) + assert ( + practice_by_id["dora_delivery_metrics_receipts"]["metrics"][ + "per_product_contract_count" + ] + == 12 + ) + assert ( + practice_by_id["dora_delivery_metrics_receipts"]["metrics"][ + "trend_values_computed" + ] + is False ) assert ( practice_by_id["slsa_build_provenance_isolation"]["status"] == "action_required" @@ -300,6 +343,19 @@ def test_gitea_capability_readback_prioritizes_professional_cicd_gaps() -> None: "https://slsa.dev/spec/v1.2/build-requirements" in payload["readback"]["market_practice_basis_refs"] ) + assert payload["delivery_performance_readback"]["source_contract_ready"] is True + assert payload["delivery_performance_readback"]["metric_count"] == 5 + assert payload["delivery_performance_readback"]["trend_values_computed"] is False + assert { + row["metric_id"] + for row in payload["delivery_performance_readback"]["metric_rows"] + } == { + "change_lead_time", + "deployment_frequency", + "failed_deployment_recovery_time", + "change_fail_rate", + "deployment_rework_rate", + } assert payload["status"] == "gitea_capability_ci_cd_optimization_ready" assert payload["ready"] is True assert payload["active_p0_gap_ids"] == [] diff --git a/apps/api/tests/test_gitea_cicd_alert_receipt_chain_readback_api.py b/apps/api/tests/test_gitea_cicd_alert_receipt_chain_readback_api.py index 56c6a1f79..46ef7d299 100644 --- a/apps/api/tests/test_gitea_cicd_alert_receipt_chain_readback_api.py +++ b/apps/api/tests/test_gitea_cicd_alert_receipt_chain_readback_api.py @@ -57,10 +57,15 @@ def test_gitea_cicd_alert_receipt_chain_readback_surfaces_receipt_gaps() -> None assert payload["schema_version"] == "gitea_cicd_alert_receipt_chain_readback_v1" assert payload["status"] == "gitea_cicd_alert_receipt_chain_gaps_present" assert payload["ready"] is False - assert payload["summary"]["receipt_chain_count"] == 5 + assert payload["summary"]["receipt_chain_count"] == 6 assert payload["summary"]["action_required_chain_count"] == 2 assert payload["summary"]["monitoring_live_receipt_gap_count"] == 60 assert payload["summary"]["full_server_dr_active_gap_count"] == 7 + assert payload["summary"]["gitea_webhook_hmac_source_ready"] is True + assert payload["summary"]["gitea_webhook_delivery_header_supported"] is True + assert ( + payload["summary"]["gitea_webhook_signature_verification_source_ready"] is True + ) assert payload["operation_boundaries"]["telegram_send_allowed"] is False assert payload["operation_boundaries"]["bot_api_call_allowed"] is False assert payload["operation_boundaries"]["backup_execution_allowed"] is False @@ -70,6 +75,21 @@ def test_gitea_cicd_alert_receipt_chain_readback_surfaces_receipt_gaps() -> None by_id = {row["chain_id"]: row for row in payload["receipt_chain_rows"]} assert by_id["gitea_cd_pipeline_notifications"]["status"] == "source_ready" + assert by_id["gitea_webhook_hmac_delivery"]["status"] == "source_ready" + assert ( + by_id["gitea_webhook_hmac_delivery"]["metrics"][ + "signature_verification_source_ready" + ] + is True + ) + assert ( + by_id["gitea_webhook_hmac_delivery"]["metrics"]["delivery_header_captured"] + is True + ) + assert ( + by_id["gitea_webhook_hmac_delivery"]["metrics"]["live_delivery_receipt_bound"] + is False + ) assert by_id["backup_notification_policy"]["status"] == "policy_ready" assert by_id["gitea_full_server_dr_receipts"]["status"] == "action_required" assert by_id["telegram_monitoring_live_receipts"]["status"] == "action_required" @@ -114,14 +134,17 @@ def test_gitea_cicd_alert_receipt_chain_keeps_restore_drill_pending_separate() - assert by_id["gitea_full_server_dr_receipts"]["status"] == ( "controlled_restore_drill_pending" ) - assert by_id["gitea_full_server_dr_receipts"]["metrics"][ - "restore_drill_only_gap" - ] is True + assert ( + by_id["gitea_full_server_dr_receipts"]["metrics"]["restore_drill_only_gap"] + is True + ) assert by_id["ai_agent_context_receipts"]["status"] == "verified_ready" @pytest.mark.asyncio -async def test_gitea_cicd_alert_receipt_chain_endpoint_returns_payload(monkeypatch) -> None: +async def test_gitea_cicd_alert_receipt_chain_endpoint_returns_payload( + monkeypatch, +) -> None: async def fake_loader() -> dict: return _payload()