wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

feat(gitea): surface professional cicd practice gaps

Browse Source
This commit is contained in:
ogt
2026-07-10 10:48:04 +08:00
parent d97b114abd
commit 82ab3770ba
2 changed files with 565 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

489
apps/api/src/services/gitea_capability_ci_cd_optimization_readback.py
View File

@@ -22,14 +22,14 @@ from src.services.backup_dr_readiness_matrix import (
from src.services.backup_notification_policy import (
load_latest_backup_notification_policy,
)
from src.services.gitea_private_inventory_p0_scorecard import (
load_latest_gitea_private_inventory_p0_scorecard,
from src.services.gitea_cicd_alert_receipt_chain_readback import (
load_latest_gitea_cicd_alert_receipt_chain_readback,
)
from src.services.gitea_full_server_backup_dr_receipts import (
load_latest_gitea_full_server_backup_dr_receipts,
)
from src.services.gitea_cicd_alert_receipt_chain_readback import (
load_latest_gitea_cicd_alert_receipt_chain_readback,
from src.services.gitea_private_inventory_p0_scorecard import (
load_latest_gitea_private_inventory_p0_scorecard,
)
from src.services.gitea_repo_bundle_backup_readback import (
load_latest_gitea_repo_bundle_backup_readback,
@@ -84,6 +84,19 @@ _REQUIRED_OPTIMIZATION_IDS = {
"dedupe_cancel_stale_run_guard",
}
_REQUIRED_PROFESSIONAL_CI_CD_PRACTICE_IDS = {
"gitea_125_actions_runner_trust",
"gitea_125_secret_variable_redaction",
"gitea_125_webhook_hmac_delivery",
"gitea_125_packages_release_registry",
"gitea_125_actions_compatibility_gap_guard",
"dora_delivery_metrics_receipts",
"slsa_build_provenance_isolation",
"gitops_argocd_automated_sync_contract",
"otel_cicd_pipeline_telemetry",
"artifact_cache_sbom_retention",
}
def load_latest_gitea_capability_ci_cd_optimization_readback(
*,
@@ -94,9 +107,7 @@ def load_latest_gitea_capability_ci_cd_optimization_readback(
root = repo_root or resolve_repo_root(Path(__file__))
operations = operations_dir or default_operations_dir(Path(__file__))
payload = build_gitea_capability_ci_cd_optimization_readback(
dev_prod_repo_readback=_load_json(
operations / _DEV_PROD_REPO_READBACK_FILE
),
dev_prod_repo_readback=_load_json(operations / _DEV_PROD_REPO_READBACK_FILE),
private_inventory=load_latest_gitea_private_inventory_p0_scorecard(),
workflow_runner_health=load_latest_gitea_workflow_runner_health(),
workflow_source_inventory=build_gitea_workflow_source_inventory(root),
@@ -128,6 +139,7 @@ def build_gitea_workflow_source_inventory(repo_root: Path) -> dict[str, Any]:
for path in workflow_files:
text = path.read_text(encoding="utf-8", errors="replace")
lowered_text = text.lower()
relative = path.relative_to(repo_root).as_posix()
secret_refs = sorted(set(re.findall(r"\bsecrets\.([A-Z0-9_]+)", text)))
vars_refs = sorted(set(re.findall(r"\bvars\.([A-Z0-9_]+)", text)))
@@ -149,10 +161,51 @@ def build_gitea_workflow_source_inventory(repo_root: Path) -> dict[str, Any]:
"vars_reference_count": vars_ref_count,
"referenced_variable_names": vars_refs,
"declared_runner_labels": _extract_runner_labels(text),
"declares_concurrency": _has_yaml_key(text, "concurrency"),
"declares_permissions": _has_yaml_key(text, "permissions"),
"declares_timeout_minutes": _has_yaml_key(text, "timeout-minutes"),
"declares_continue_on_error": _has_yaml_key(text, "continue-on-error"),
"declares_environment": _has_yaml_key(text, "environment"),
"uses_otel": "otel_" in lowered_text or "opentelemetry" in lowered_text,
"uses_artifact_transfer": (
"upload-artifact" in lowered_text
or "download-artifact" in lowered_text
),
"uses_cache_volume_or_action": (
"actions/cache" in lowered_text
or "venv-cache" in lowered_text
or "npm-cache" in lowered_text
or "pip-cache" in lowered_text
),
"uses_sbom_or_provenance": any(
marker in lowered_text
for marker in ("sbom", "provenance", "cosign", "syft", "trivy")
),
"uses_registry_or_package_channel": (
"registry.wooo.work" in text
or "harbor" in lowered_text
or "/packages/" in lowered_text
),
"uses_argocd_gitops": "argocd" in lowered_text,
"uses_notification_or_webhook": (
"telegram" in lowered_text or "webhook" in lowered_text
),
}
)
workflow_ids = sorted(row["workflow_id"] for row in rows)
gitea_ignored_syntax_keys = {
"declares_concurrency",
"declares_permissions",
"declares_timeout_minutes",
"declares_continue_on_error",
"declares_environment",
}
gitea_ignored_syntax_workflow_ids = sorted(
row["workflow_id"]
for row in rows
if any(row.get(key) is True for key in gitea_ignored_syntax_keys)
)
return {
"source_ref": ".gitea/workflows",
"source_present": workflow_dir.is_dir(),
@@ -210,6 +263,27 @@ def build_gitea_workflow_source_inventory(repo_root: Path) -> dict[str, Any]:
"has_agent_market_watch": "agent-market-watch" in workflow_ids,
"has_ai_technology_watch": "ai-technology-watch" in workflow_ids,
},
"professional_feature_rollups": {
"gitea_ignored_syntax_workflow_count": len(
gitea_ignored_syntax_workflow_ids
),
"gitea_ignored_syntax_workflow_ids": gitea_ignored_syntax_workflow_ids,
"otel_workflow_count": _count_where(rows, "uses_otel"),
"artifact_transfer_workflow_count": _count_where(
rows, "uses_artifact_transfer"
),
"cache_workflow_count": _count_where(rows, "uses_cache_volume_or_action"),
"sbom_or_provenance_workflow_count": _count_where(
rows, "uses_sbom_or_provenance"
),
"registry_or_package_channel_workflow_count": _count_where(
rows, "uses_registry_or_package_channel"
),
"argocd_gitops_workflow_count": _count_where(rows, "uses_argocd_gitops"),
"notification_or_webhook_workflow_count": _count_where(
rows, "uses_notification_or_webhook"
),
},
}
@@ -248,6 +322,9 @@ 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"))
professional_feature_rollups = _dict(
workflow_source_inventory.get("professional_feature_rollups")
)
full_dr_non_restore_gap_ids = [
gap for gap in full_dr_gap_ids if gap != "restore_drill_approval_gate"
]
@@ -257,7 +334,9 @@ def build_gitea_capability_ci_cd_optimization_readback(
)
full_dr_verified_ready = full_dr.get("ready") is True
all_repos_have_refs = dev_prod_summary.get("all_expected_product_repos_have_ssh_refs") is True
all_repos_have_refs = (
dev_prod_summary.get("all_expected_product_repos_have_ssh_refs") is True
)
all_repos_have_dev_main = (
dev_prod_summary.get("all_expected_product_repos_have_dev_and_main") is True
)
@@ -335,7 +414,9 @@ def build_gitea_capability_ci_cd_optimization_readback(
],
"preserve SSH/authenticated truth before creating or replacing repos",
{
"public_visible_count": _int(dev_prod_summary.get("public_visible_count")),
"public_visible_count": _int(
dev_prod_summary.get("public_visible_count")
),
"private_or_auth_only_count": _int(
dev_prod_summary.get("private_or_auth_only_count")
),
@@ -345,7 +426,9 @@ def build_gitea_capability_ci_cd_optimization_readback(
"pull_request_code_review",
"collaboration",
"Pull requests and code review",
"partially_ready" if known_workflows.get("has_code_review") else "action_required",
"partially_ready"
if known_workflows.get("has_code_review")
else "action_required",
"code-review workflow exists; repository PR policy still needs live inventory",
[
".gitea/workflows/code-review.yaml",
@@ -741,22 +824,294 @@ def build_gitea_capability_ci_cd_optimization_readback(
),
_optimization(
"dedupe_cancel_stale_run_guard",
"source_ready" if known_workflows.get("has_cd_pipeline") else "design_required",
"source_ready"
if known_workflows.get("has_cd_pipeline")
else "design_required",
"Implement Gitea-compatible stale-run dedupe guards",
"CD source includes stale-run guard; live receipt wiring is the remaining work"
if known_workflows.get("has_cd_pipeline")
else "Gitea Actions compatibility requires workflow/runner-side guards for some controls",
[".gitea/workflows/cd.yaml", "ops/runner/read-public-gitea-actions-queue.py"],
[
".gitea/workflows/cd.yaml",
"ops/runner/read-public-gitea-actions-queue.py",
],
"wire stale-run guard receipts into the Gitea capability readback"
if known_workflows.get("has_cd_pipeline")
else "add lease/marker based cancel-stale semantics instead of relying on unsupported syntax",
p0=True,
),
]
professional_practices = [
_professional_practice(
"gitea_125_actions_runner_trust",
"Gitea 1.25 Actions",
"verified_ready"
if workflow_source_count > 0 and not runner_action_ids
else "action_required",
"Trusted act runners, explicit labels, and non-110 isolation",
"workflow source and runner-label contracts are aligned"
if workflow_source_count > 0 and not runner_action_ids
else "runner trust and label ownership need closure before expansion",
[
".gitea/workflows",
"GET /api/v1/agents/gitea-workflow-runner-health",
"https://docs.gitea.com/1.25/usage/actions/overview",
"https://docs.gitea.com/1.25/usage/actions/act-runner",
],
"keep trusted labels and runner health fresh; evaluate ephemeral runners for broader shared-runner lanes",
{
"workflow_source_file_count": workflow_source_count,
"runner_contracts_requiring_action": runner_action_ids,
},
p0=True,
),
_professional_practice(
"gitea_125_secret_variable_redaction",
"Gitea 1.25 Actions",
"source_ready"
if _int(secret_rollups.get("secret_reference_count")) > 0
and secret_rollups.get("secret_values_collected") is False
else "authenticated_inventory_required",
"Secret and variable metadata inventory without value collection",
"workflow references are inventoried and secret values remain unread"
if _int(secret_rollups.get("secret_reference_count")) > 0
else "configured Gitea secret and variable names still need redacted inventory",
[
".gitea/workflows",
"GET /api/v1/agents/gitea-actions-secret-variable-name-inventory",
"https://docs.gitea.com/1.25/usage/actions/secrets",
"https://docs.gitea.com/1.25/usage/actions/actions-variables",
],
"verify presence and scope using redacted name-only metadata; never read secret values",
secret_rollups,
p0=True,
),
_professional_practice(
"gitea_125_webhook_hmac_delivery",
"Gitea 1.25 Repository",
"source_ready"
if quiet_notification_count > 0 and alert_chain_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
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",
{
"quiet_notification_policy_count": quiet_notification_count,
"alert_receipt_chain_ready": alert_chain_ready,
},
),
_professional_practice(
"gitea_125_packages_release_registry",
"Gitea 1.25 Packages",
"controlled_apply_candidate"
if _int(
professional_feature_rollups.get(
"registry_or_package_channel_workflow_count"
)
)
> 0
else "action_required",
"Package registry, release artifacts, and rollback artifacts",
"registry channel is present, but Gitea package/release promotion is not standardized"
if _int(
professional_feature_rollups.get(
"registry_or_package_channel_workflow_count"
)
)
> 0
else "package and release artifact channel is missing from workflow source",
[
".gitea/workflows/cd.yaml",
"GET /api/v1/agents/package-supply-chain-inventory",
"https://docs.gitea.com/1.25/usage/packages/overview",
],
"standardize package/release promotion, rollback artifact naming, and package event receipts per product",
{
"registry_or_package_channel_workflow_count": _int(
professional_feature_rollups.get(
"registry_or_package_channel_workflow_count"
)
)
},
),
_professional_practice(
"gitea_125_actions_compatibility_gap_guard",
"Gitea 1.25 Actions",
"source_ready"
if known_workflows.get("has_cd_pipeline")
and _int(
professional_feature_rollups.get("gitea_ignored_syntax_workflow_count")
)
> 0
else "implemented",
"Gitea-compatible guards for ignored Actions syntax",
"workflow source detects ignored syntax and uses lease/marker/readback guards for critical CD semantics"
if _int(
professional_feature_rollups.get("gitea_ignored_syntax_workflow_count")
)
> 0
else "no ignored syntax detected in committed workflows",
[
".gitea/workflows/cd.yaml",
"ops/runner/read-public-gitea-actions-queue.py",
"https://docs.gitea.com/1.25/usage/actions/comparison",
],
"keep unsupported syntax detection visible and prefer Gitea-compatible receipt guards over copied YAML semantics",
{
"gitea_ignored_syntax_workflow_count": _int(
professional_feature_rollups.get(
"gitea_ignored_syntax_workflow_count"
)
),
"gitea_ignored_syntax_workflow_ids": _strings(
professional_feature_rollups.get(
"gitea_ignored_syntax_workflow_ids"
)
),
},
p0=True,
),
_professional_practice(
"dora_delivery_metrics_receipts",
"DORA",
"action_required",
"Five-metric delivery performance readback",
"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",
),
_professional_practice(
"slsa_build_provenance_isolation",
"OpenSSF SLSA",
"action_required"
if _int(
professional_feature_rollups.get("sbom_or_provenance_workflow_count")
)
== 0
else "source_ready",
"Build provenance, digest identity, and isolated build controls",
"workflow source does not yet emit SLSA provenance/SBOM receipts"
if _int(
professional_feature_rollups.get("sbom_or_provenance_workflow_count")
)
== 0
else "workflow source contains provenance/SBOM markers",
[
"GET /api/v1/agents/package-supply-chain-inventory",
"https://slsa.dev/spec/v1.2/build-requirements",
],
"add digest-bound provenance/SBOM output and runner isolation evidence to build artifacts",
{
"sbom_or_provenance_workflow_count": _int(
professional_feature_rollups.get(
"sbom_or_provenance_workflow_count"
)
)
},
),
_professional_practice(
"gitops_argocd_automated_sync_contract",
"GitOps / Argo CD",
"source_ready"
if _int(professional_feature_rollups.get("argocd_gitops_workflow_count"))
> 0
else "action_required",
"Git commit driven deployment and Argo CD sync/readback",
"CD workflow uses ArgoCD GitOps and deploy-marker readback",
[
".gitea/workflows/cd.yaml",
"GET /api/v1/agents/p0-cicd-baseline-source-readiness",
"https://argo-cd.readthedocs.io/en/stable/user-guide/auto_sync/",
],
"keep Argo sync, health, revision, deploy marker, and smoke readback tied to the same commit",
{
"argocd_gitops_workflow_count": _int(
professional_feature_rollups.get("argocd_gitops_workflow_count")
)
},
p0=True,
),
_professional_practice(
"otel_cicd_pipeline_telemetry",
"OpenTelemetry",
"source_ready"
if _int(professional_feature_rollups.get("otel_workflow_count")) > 0
else "action_required",
"CI/CD pipeline telemetry and low-cardinality result tags",
"workflow source exports OTEL service/resource attributes for CD lanes",
[
".gitea/workflows/cd.yaml",
".gitea/workflows/cd-dev.yaml",
".gitea/workflows/e2e-health.yaml",
"https://opentelemetry.io/docs/specs/semconv/cicd/cicd-spans/",
],
"bind pipeline result, run id, commit, environment, and error type into CI/CD span/readback receipts",
{
"otel_workflow_count": _int(
professional_feature_rollups.get("otel_workflow_count")
)
},
),
_professional_practice(
"artifact_cache_sbom_retention",
"CI/CD Artifact Governance",
"action_required"
if _int(
professional_feature_rollups.get("artifact_transfer_workflow_count")
)
== 0
or _int(
professional_feature_rollups.get("sbom_or_provenance_workflow_count")
)
== 0
else "source_ready",
"Cache keys, artifact retention, SBOM, and receipt retention",
"workflow source still lacks unified artifact/SBOM retention receipts"
if _int(
professional_feature_rollups.get("artifact_transfer_workflow_count")
)
== 0
or _int(
professional_feature_rollups.get("sbom_or_provenance_workflow_count")
)
== 0
else "workflow source contains artifact and SBOM/provenance markers",
[
".gitea/workflows",
"GET /api/v1/agents/gitea-workflow-runner-health",
],
"standardize artifact upload, cache keys, SBOM/provenance naming, and retention windows across workflows",
{
"artifact_transfer_workflow_count": _int(
professional_feature_rollups.get("artifact_transfer_workflow_count")
),
"cache_workflow_count": _int(
professional_feature_rollups.get("cache_workflow_count")
),
"sbom_or_provenance_workflow_count": _int(
professional_feature_rollups.get(
"sbom_or_provenance_workflow_count"
)
),
},
),
]
rollups = _build_rollups(
capabilities=capabilities,
optimizations=optimizations,
professional_practices=professional_practices,
dev_prod_summary=dev_prod_summary,
workflow_source_count=workflow_source_count,
workflow_contract_count=workflow_contract_count,
@@ -820,17 +1175,26 @@ def build_gitea_capability_ci_cd_optimization_readback(
"official_gitea_feature_refs": [
"https://docs.gitea.com/1.25/usage/access-control/permissions",
"https://docs.gitea.com/1.25/usage/actions/overview",
"https://docs.gitea.com/1.25/usage/actions/act-runner",
"https://docs.gitea.com/1.25/usage/actions/comparison",
"https://docs.gitea.com/1.25/usage/actions/secrets",
"https://docs.gitea.com/1.25/usage/actions/actions-variables",
"https://docs.gitea.com/1.25/usage/packages/overview",
"https://docs.gitea.com/1.25/usage/repository/webhooks",
],
"market_practice_basis_refs": [
"https://dora.dev/guides/dora-metrics/",
"https://slsa.dev/spec/v1.2/build-requirements",
"https://argo-cd.readthedocs.io/en/stable/user-guide/auto_sync/",
"https://opentelemetry.io/docs/specs/semconv/cicd/cicd-spans/",
],
"safe_next_actions": _safe_next_actions(capabilities, optimizations),
},
"rollups": rollups,
"workflow_source_inventory": workflow_source_inventory,
"capability_rows": capabilities,
"ci_cd_optimization_rows": optimizations,
"professional_ci_cd_practice_rows": professional_practices,
}
_require_payload(payload, "gitea_capability_ci_cd_optimization_readback")
return payload
@@ -884,10 +1248,36 @@ def _optimization(
}
def _professional_practice(
practice_id: str,
framework: str,
status_value: str,
title: str,
usage_state: str,
evidence_refs: list[str],
next_action: str,
metrics: dict[str, Any] | None = None,
*,
p0: bool = False,
) -> dict[str, Any]:
return {
"id": practice_id,
"framework": framework,
"status": status_value,
"priority": "P0" if p0 else "P1",
"title": title,
"usage_state": usage_state,
"evidence_refs": evidence_refs,
"next_action": next_action,
"metrics": metrics or {},
}
def _build_rollups(
*,
capabilities: list[dict[str, Any]],
optimizations: list[dict[str, Any]],
professional_practices: list[dict[str, Any]],
dev_prod_summary: dict[str, Any],
workflow_source_count: int,
workflow_contract_count: int,
@@ -895,8 +1285,28 @@ def _build_rollups(
) -> dict[str, Any]:
capability_status = _count_by(capabilities, "status")
optimization_status = _count_by(optimizations, "status")
professional_status = _count_by(professional_practices, "status")
p0_capabilities = [row for row in capabilities if row.get("priority") == "P0"]
p0_optimizations = [row for row in optimizations if row.get("priority") == "P0"]
p0_professional_practices = [
row for row in professional_practices if row.get("priority") == "P0"
]
ready_statuses = {"verified_ready", "implemented", "source_ready"}
capability_gap_ids = [
str(row.get("id"))
for row in capabilities
if row.get("status") not in ready_statuses
]
optimization_gap_ids = [
str(row.get("id"))
for row in optimizations
if row.get("status") not in ready_statuses
]
professional_gap_ids = [
str(row.get("id"))
for row in professional_practices
if row.get("status") not in ready_statuses
]
return {
"capability_count": len(capabilities),
"capability_status_counts": capability_status,
@@ -917,6 +1327,21 @@ def _build_rollups(
),
"p0_optimization_count": len(p0_optimizations),
"p0_optimization_gap_count": _gap_count(p0_optimizations),
"professional_ci_cd_practice_count": len(professional_practices),
"professional_ci_cd_practice_status_counts": professional_status,
"professional_ci_cd_practice_gap_count": len(professional_gap_ids),
"professional_ci_cd_practice_gap_ids": professional_gap_ids,
"p0_professional_ci_cd_practice_count": len(p0_professional_practices),
"p0_professional_ci_cd_practice_gap_count": _gap_count(
p0_professional_practices
),
"market_professional_ci_cd_gap_count": len(professional_gap_ids),
"gitea_feature_utilization_gap_ids": sorted(
set(capability_gap_ids + optimization_gap_ids + professional_gap_ids)
),
"all_gitea_features_fully_utilized": not (
capability_gap_ids or optimization_gap_ids or professional_gap_ids
),
"expected_product_count": _int(dev_prod_summary.get("expected_product_count")),
"ssh_verified_product_repo_count": _int(
dev_prod_summary.get("ssh_repo_present_count")
@@ -939,13 +1364,9 @@ def _safe_next_actions(
row
for row in capabilities + optimizations
if row.get("priority") == "P0"
and row.get("status")
not in {"verified_ready", "implemented", "source_ready"}
]
return [
f"{row['id']}: {row['next_action']}"
for row in rows[:8]
and row.get("status") not in {"verified_ready", "implemented", "source_ready"}
]
return [f"{row['id']}: {row['next_action']}" for row in rows[:8]]
def _require_payload(payload: dict[str, Any], label: str) -> None:
@@ -973,26 +1394,45 @@ def _require_payload(payload: dict[str, Any], label: str) -> None:
flag for flag in blocked_true_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}"
)
capabilities = _list(payload.get("capability_rows"))
optimizations = _list(payload.get("ci_cd_optimization_rows"))
capability_ids = {str(row.get("id") or "") for row in capabilities if isinstance(row, dict)}
professional_practices = _list(payload.get("professional_ci_cd_practice_rows"))
capability_ids = {
str(row.get("id") or "") for row in capabilities if isinstance(row, dict)
}
optimization_ids = {
str(row.get("id") or "") for row in optimizations if isinstance(row, dict)
}
professional_practice_ids = {
str(row.get("id") or "")
for row in professional_practices
if isinstance(row, dict)
}
missing_capabilities = sorted(_REQUIRED_CAPABILITY_IDS - capability_ids)
missing_optimizations = sorted(_REQUIRED_OPTIMIZATION_IDS - optimization_ids)
missing_professional_practices = sorted(
_REQUIRED_PROFESSIONAL_CI_CD_PRACTICE_IDS - professional_practice_ids
)
if missing_capabilities:
raise ValueError(f"{label}: missing capability rows: {missing_capabilities}")
if missing_optimizations:
raise ValueError(f"{label}: missing optimization rows: {missing_optimizations}")
if missing_professional_practices:
raise ValueError(
f"{label}: missing professional CI/CD practice rows: {missing_professional_practices}"
)
rollups = _dict(payload.get("rollups"))
if rollups.get("capability_count") != len(capabilities):
raise ValueError(f"{label}: capability_count mismatch")
if rollups.get("ci_cd_optimization_count") != len(optimizations):
raise ValueError(f"{label}: ci_cd_optimization_count mismatch")
if rollups.get("professional_ci_cd_practice_count") != len(professional_practices):
raise ValueError(f"{label}: professional_ci_cd_practice_count mismatch")
if rollups.get("capability_status_counts") != _count_by(capabilities, "status"):
raise ValueError(f"{label}: capability_status_counts mismatch")
if rollups.get("ci_cd_optimization_status_counts") != _count_by(
@@ -1000,6 +1440,11 @@ def _require_payload(payload: dict[str, Any], label: str) -> None:
"status",
):
raise ValueError(f"{label}: ci_cd_optimization_status_counts mismatch")
if rollups.get("professional_ci_cd_practice_status_counts") != _count_by(
professional_practices,
"status",
):
raise ValueError(f"{label}: professional_ci_cd_practice_status_counts mismatch")
_require_no_plaintext_secret_keys(payload, label)
@@ -1018,7 +1463,9 @@ def _require_no_plaintext_secret_keys(value: Any, label: str, path: str = "$") -
for key, child in value.items():
lowered = str(key).lower()
if lowered in forbidden or lowered.endswith("_secret_value"):
raise ValueError(f"{label}: forbidden plaintext-like key at {path}.{key}")
raise ValueError(
f"{label}: forbidden plaintext-like key at {path}.{key}"
)
_require_no_plaintext_secret_keys(child, label, f"{path}.{key}")
elif isinstance(value, list):
for index, child in enumerate(value):

128
apps/api/tests/test_gitea_capability_ci_cd_optimization_readback_api.py
View File

@@ -1,5 +1,6 @@
from __future__ import annotations
# ruff: noqa: E402
import json
import os
from pathlib import Path
@@ -140,8 +141,7 @@ def test_gitea_capability_readback_rolls_up_complete_feature_surface() -> None:
payload = _readback_payload()
assert (
payload["schema_version"]
== "gitea_capability_ci_cd_optimization_readback_v1"
payload["schema_version"] == "gitea_capability_ci_cd_optimization_readback_v1"
)
assert payload["source_control_authority"] == "gitea"
assert payload["status"] == "gitea_capability_ci_cd_optimization_ready"
@@ -149,12 +149,29 @@ def test_gitea_capability_readback_rolls_up_complete_feature_surface() -> None:
assert payload["rollups"]["ssh_verified_product_repo_count"] == 12
assert payload["rollups"]["dev_prod_environment_split_ready_count"] == 12
assert payload["workflow_source_inventory"]["workflow_file_count"] == 12
assert payload["workflow_source_inventory"]["secret_variable_rollups"][
"secret_values_collected"
] is False
assert payload["workflow_source_inventory"]["secret_variable_rollups"][
"secret_reference_count"
] > 0
professional_feature_rollups = payload["workflow_source_inventory"][
"professional_feature_rollups"
]
assert professional_feature_rollups["gitea_ignored_syntax_workflow_count"] == 9
assert professional_feature_rollups["otel_workflow_count"] == 3
assert professional_feature_rollups["argocd_gitops_workflow_count"] == 1
assert (
professional_feature_rollups["registry_or_package_channel_workflow_count"] == 4
)
assert professional_feature_rollups["sbom_or_provenance_workflow_count"] == 0
assert professional_feature_rollups["artifact_transfer_workflow_count"] == 0
assert (
payload["workflow_source_inventory"]["secret_variable_rollups"][
"secret_values_collected"
]
is False
)
assert (
payload["workflow_source_inventory"]["secret_variable_rollups"][
"secret_reference_count"
]
> 0
)
assert (
len(
payload["workflow_source_inventory"]["secret_variable_rollups"][
@@ -166,22 +183,27 @@ def test_gitea_capability_readback_rolls_up_complete_feature_surface() -> None:
capability_by_id = {row["id"]: row for row in payload["capability_rows"]}
assert capability_by_id["code_git_source_control"]["status"] == "verified_ready"
assert (
capability_by_id["dev_prod_environment_split"]["status"]
== "verified_ready"
)
assert capability_by_id["dev_prod_environment_split"]["status"] == "verified_ready"
assert capability_by_id["gitea_actions_workflows"]["status"] == "verified_ready"
assert capability_by_id["gitea_actions_workflows"]["metrics"] == {
"workflow_source_file_count": 12,
"workflow_contract_count": 12,
}
assert capability_by_id["gitea_actions_runners"]["status"] == "verified_ready"
assert capability_by_id["branch_protection_required_checks"]["status"] == "source_ready"
assert (
capability_by_id["branch_protection_required_checks"]["status"]
== "source_ready"
)
assert capability_by_id["backup_dr_restore"]["status"] == "source_ready"
assert capability_by_id["backup_dr_restore"]["metrics"][
"gitea_full_dr_source_ready_component_count"
] == 6
assert capability_by_id["observability_alert_receipts"]["status"] == "verified_ready"
assert (
capability_by_id["backup_dr_restore"]["metrics"][
"gitea_full_dr_source_ready_component_count"
]
== 6
)
assert (
capability_by_id["observability_alert_receipts"]["status"] == "verified_ready"
)
assert capability_by_id["observability_alert_receipts"]["metrics"] == {
"alert_receipt_chain_ready": True,
"alert_receipt_chain_status": "gitea_cicd_alert_receipt_chain_ready",
@@ -198,29 +220,22 @@ def test_gitea_capability_readback_rolls_up_complete_feature_surface() -> None:
def test_gitea_capability_readback_prioritizes_professional_cicd_gaps() -> None:
payload = _readback_payload()
optimization_by_id = {
row["id"]: row for row in payload["ci_cd_optimization_rows"]
}
optimization_by_id = {row["id"]: row for row in payload["ci_cd_optimization_rows"]}
assert (
optimization_by_id["gitea_actions_source_authority"]["status"]
== "implemented"
optimization_by_id["gitea_actions_source_authority"]["status"] == "implemented"
)
assert (
optimization_by_id["dev_prod_branch_promotion_contract"]["status"]
== "implemented"
)
assert (
optimization_by_id["workflow_contract_freshness"]["status"]
== "implemented"
)
assert optimization_by_id["workflow_contract_freshness"]["status"] == "implemented"
assert (
optimization_by_id["runner_label_isolation_attestation"]["status"]
== "implemented"
)
assert (
optimization_by_id["dedupe_cancel_stale_run_guard"]["status"]
== "source_ready"
optimization_by_id["dedupe_cancel_stale_run_guard"]["status"] == "source_ready"
)
assert (
optimization_by_id["branch_protection_required_checks"]["status"]
@@ -234,6 +249,56 @@ def test_gitea_capability_readback_prioritizes_professional_cicd_gaps() -> None:
payload["ci_cd_optimization_rows"]
)
assert payload["rollups"]["p0_optimization_gap_count"] == 0
assert payload["rollups"]["professional_ci_cd_practice_count"] == len(
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"]["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",
]
practice_by_id = {
row["id"]: row for row in payload["professional_ci_cd_practice_rows"]
}
assert (
practice_by_id["gitea_125_actions_runner_trust"]["status"] == "verified_ready"
)
assert (
practice_by_id["gitea_125_secret_variable_redaction"]["status"]
== "source_ready"
)
assert (
practice_by_id["gitea_125_actions_compatibility_gap_guard"]["status"]
== "source_ready"
)
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"] == "action_required"
)
assert (
practice_by_id["slsa_build_provenance_isolation"]["status"] == "action_required"
)
assert (
practice_by_id["artifact_cache_sbom_retention"]["status"] == "action_required"
)
assert (
"https://dora.dev/guides/dora-metrics/"
in payload["readback"]["market_practice_basis_refs"]
)
assert (
"https://slsa.dev/spec/v1.2/build-requirements"
in payload["readback"]["market_practice_basis_refs"]
)
assert payload["status"] == "gitea_capability_ci_cd_optimization_ready"
assert payload["ready"] is True
assert payload["active_p0_gap_ids"] == []
@@ -250,12 +315,13 @@ def test_gitea_capability_readback_endpoint_returns_payload(monkeypatch) -> None
app.include_router(router, prefix="/api/v1")
client = TestClient(app)
response = client.get(
"/api/v1/agents/gitea-capability-ci-cd-optimization-readback"
)
response = client.get("/api/v1/agents/gitea-capability-ci-cd-optimization-readback")
assert response.status_code == 200
data = response.json()
assert data["schema_version"] == "gitea_capability_ci_cd_optimization_readback_v1"
assert data["rollups"]["capability_count"] == len(data["capability_rows"])
assert data["rollups"]["professional_ci_cd_practice_count"] == len(
data["professional_ci_cd_practice_rows"]
)
assert data["rollups"]["workflow_contract_fresh"] is True