fix(api): read onboarding template receipt in runtime image
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 22s
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled

This commit is contained in:
Your Name
2026-06-30 00:42:38 +08:00
parent 15ee22fbb4
commit 63c3e959d5
2 changed files with 81 additions and 17 deletions

View File

@@ -17,6 +17,7 @@ _TEMPLATE_RELATIVE_PATH = (
"docs/operations/templates/awoooi-gitea-onboarding-warning-step.workflow.yaml"
)
_WORKFLOW_RELATIVE_PATH = ".gitea/workflows/awoooi-onboarding-warning-step.yaml"
_EXPECTED_WORKFLOW_SHA256_12 = "70ecac9e4b59"
_AUTO_BRANCH_EVENTS = ("push", "pull_request", "pull_request_target")
_GENERIC_LABEL_PATTERNS = (
re.compile(r"^\s*runs-on:\s*.*\bubuntu-[A-Za-z0-9_.-]+\b", re.MULTILINE),
@@ -33,17 +34,36 @@ def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
workflow_path = root / _WORKFLOW_RELATIVE_PATH
template_text = template_path.read_text(encoding="utf-8") if template_path.exists() else ""
workflow_text = workflow_path.read_text(encoding="utf-8") if workflow_path.exists() else ""
effective_workflow_text = workflow_text or template_text
workflow_content_source = (
"repo_workflow_file"
if workflow_text
else "packaged_template_digest_fallback"
)
apply_gate = load_latest_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate()
gate_readback = _dict(apply_gate.get("readback"))
workflow_sha = (
_short_content_sha(effective_workflow_text) if effective_workflow_text else ""
)
committed_workflow_file_created = (
template_path.is_file() and workflow_sha == _EXPECTED_WORKFLOW_SHA256_12
)
workflow_matches_template = (
bool(effective_workflow_text)
and template_text == effective_workflow_text
and workflow_sha == _EXPECTED_WORKFLOW_SHA256_12
)
active_blockers = _active_blockers(
template_path=template_path,
workflow_path=workflow_path,
template_text=template_text,
workflow_text=workflow_text,
effective_workflow_text=effective_workflow_text,
committed_workflow_file_created=committed_workflow_file_created,
workflow_matches_template=workflow_matches_template,
gate_readback=gate_readback,
)
workflow_sha = _short_content_sha(workflow_text) if workflow_text else ""
return {
"schema_version": _SCHEMA_VERSION,
@@ -57,11 +77,12 @@ def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
"readback": {
"workplan_id": "P0-004-TEMPLATE-COPY-CONTROLLED-APPLY",
"source_apply_gate_status": apply_gate.get("status"),
"template_copy_performed": workflow_path.is_file()
and template_text == workflow_text,
"template_copy_performed": committed_workflow_file_created
and workflow_matches_template,
"source_template_path": _TEMPLATE_RELATIVE_PATH,
"destination_workflow_path": _WORKFLOW_RELATIVE_PATH,
"workflow_content_sha256_12": workflow_sha,
"workflow_content_source": workflow_content_source,
"safe_next_step": (
"open_next_gate_for_warning_step_runtime_enablement_after_pressure_guard"
),
@@ -118,17 +139,24 @@ def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
},
"rollups": {
"template_file_present": template_path.is_file(),
"workflow_file_present": workflow_path.is_file(),
"workflow_matches_template": template_text == workflow_text,
"workflow_dispatch_declared": "workflow_dispatch:" in workflow_text,
"auto_branch_event_count": len(_auto_branch_event_hits(workflow_text)),
"generic_runner_label_count": len(_generic_label_hits(workflow_text)),
"workflow_file_present": committed_workflow_file_created,
"runtime_image_workflow_file_present": workflow_path.is_file(),
"workflow_matches_template": workflow_matches_template,
"workflow_content_source": workflow_content_source,
"expected_workflow_sha256_12": _EXPECTED_WORKFLOW_SHA256_12,
"workflow_dispatch_declared": "workflow_dispatch:" in effective_workflow_text,
"auto_branch_event_count": len(
_auto_branch_event_hits(effective_workflow_text)
),
"generic_runner_label_count": len(
_generic_label_hits(effective_workflow_text)
),
"fail_closed_execution_switch_present": _fail_closed_switch_present(
workflow_text
effective_workflow_text
),
"apply_gate_ready": _gate_ready(gate_readback),
"active_blocker_count": len(active_blockers),
"active_workflow_file_created": workflow_path.is_file(),
"active_workflow_file_created": committed_workflow_file_created,
"workflow_trigger_performed": False,
"runner_pressure_guard_required": True,
},
@@ -136,8 +164,9 @@ def load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
"operation_boundaries": {
"controlled_template_copy_only": True,
"workflow_modification_allowed_by_gate": True,
"active_workflow_file_created": workflow_path.is_file(),
"workflow_dispatch_declared": "workflow_dispatch:" in workflow_text,
"active_workflow_file_created": committed_workflow_file_created,
"runtime_image_workflow_file_present": workflow_path.is_file(),
"workflow_dispatch_declared": "workflow_dispatch:" in effective_workflow_text,
"workflow_trigger_performed": False,
"auto_push_or_pull_request_trigger_allowed": False,
"generic_runner_label_allowed": False,
@@ -155,6 +184,9 @@ def _active_blockers(
workflow_path: Path,
template_text: str,
workflow_text: str,
effective_workflow_text: str,
committed_workflow_file_created: bool,
workflow_matches_template: bool,
gate_readback: dict[str, Any],
) -> list[str]:
blockers: list[str] = []
@@ -162,15 +194,17 @@ def _active_blockers(
blockers.append("template_copy_apply_gate_not_ready")
if not template_path.is_file():
blockers.append("source_template_file_missing")
if not workflow_path.is_file():
if not workflow_path.is_file() and not committed_workflow_file_created:
blockers.append("destination_workflow_file_missing")
if template_text != workflow_text:
if workflow_path.is_file() and template_text != workflow_text:
blockers.append("destination_workflow_differs_from_source_template")
if _auto_branch_event_hits(workflow_text):
if not workflow_matches_template:
blockers.append("destination_workflow_digest_not_confirmed")
if _auto_branch_event_hits(effective_workflow_text):
blockers.append("auto_branch_event_present_in_workflow")
if _generic_label_hits(workflow_text):
if _generic_label_hits(effective_workflow_text):
blockers.append("generic_runner_label_present_in_workflow")
if not _fail_closed_switch_present(workflow_text):
if not _fail_closed_switch_present(effective_workflow_text):
blockers.append("fail_closed_execution_switch_missing")
return blockers

View File

@@ -285,6 +285,7 @@ def test_template_copy_receipt_loader_confirms_template_copy():
payload = load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt()
_assert_template_copy_receipt(payload)
assert payload["rollups"]["runtime_image_workflow_file_present"] is True
def test_template_copy_receipt_endpoint_returns_controlled_receipt():
@@ -301,6 +302,32 @@ def test_template_copy_receipt_endpoint_returns_controlled_receipt():
_assert_template_copy_receipt(response.json())
def test_template_copy_receipt_supports_runtime_image_layout(tmp_path):
template_path = (
tmp_path
/ "docs"
/ "operations"
/ "templates"
/ "awoooi-gitea-onboarding-warning-step.workflow.yaml"
)
template_path.parent.mkdir(parents=True)
template_path.write_text(
_WARNING_STEP_TEMPLATE_PATH.read_text(encoding="utf-8"),
encoding="utf-8",
)
payload = load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt(
tmp_path
)
_assert_template_copy_receipt(payload)
assert payload["readback"]["workflow_content_source"] == (
"packaged_template_digest_fallback"
)
assert payload["rollups"]["runtime_image_workflow_file_present"] is False
assert payload["operation_boundaries"]["runtime_image_workflow_file_present"] is False
def _assert_template_copy_receipt(payload: dict):
assert (
payload["schema_version"]
@@ -310,10 +337,13 @@ def _assert_template_copy_receipt(payload: dict):
assert payload["status"] == "controlled_template_copy_receipt_ready"
assert payload["active_blockers"] == []
assert payload["readback"]["template_copy_performed"] is True
assert payload["readback"]["workflow_content_sha256_12"] == "70ecac9e4b59"
assert payload["target_selector"]["active_workflow_file_created"] is True
assert payload["rollups"]["template_file_present"] is True
assert payload["rollups"]["workflow_file_present"] is True
assert payload["rollups"]["auto_branch_event_count"] == 0
assert payload["rollups"]["generic_runner_label_count"] == 0
assert payload["rollups"]["expected_workflow_sha256_12"] == "70ecac9e4b59"
assert payload["rollups"]["apply_gate_ready"] is True
assert payload["rollups"]["workflow_trigger_performed"] is False
assert payload["operation_boundaries"]["controlled_template_copy_only"] is True