Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
202 changes: 202 additions & 0 deletions .github/utils/test_release_image_cache_sync_contract.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,202 @@
#!/usr/bin/env python3
import os
import subprocess
import tempfile
import unittest
from pathlib import Path

import yaml

ROOT = Path(__file__).resolve().parents[2]
WORKFLOW = ROOT / ".github/workflows/release-image-cache-sync.yml"
CONTRACT_WORKFLOW = (
ROOT / ".github/workflows/test-release-image-cache-sync-contract.yml"
)
CONTRACT_PATHS = [
".github/workflows/release-image-cache-sync.yml",
".github/workflows/test-release-image-cache-sync-contract.yml",
".github/utils/test_release_image_cache_sync_contract.py",
]
VALID_DIGEST_A = "sha256:" + "a" * 64
VALID_DIGEST_B = "sha256:" + "b" * 64


def load_workflow(path: Path) -> dict:
workflow = yaml.safe_load(path.read_text())
workflow_on = workflow.get("on", workflow.get(True))
if not isinstance(workflow_on, dict):
raise AssertionError("workflow must define an on mapping")
workflow["on"] = workflow_on
return workflow


def step_by_id(workflow: dict, step_id: str) -> dict:
steps = workflow["jobs"]["release-image"]["steps"]
matches = [step for step in steps if step.get("id") == step_id]
if len(matches) != 1:
raise AssertionError(f"expected one step with id {step_id}, got {len(matches)}")
return matches[0]


class ReleaseImageCacheSyncContractTest(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.workflow = load_workflow(WORKFLOW)
cls.contract_workflow = load_workflow(CONTRACT_WORKFLOW)
cls.selector = step_by_id(cls.workflow, "select_image_index_digest")

def test_contract_workflow_trigger_matrix_is_exact(self) -> None:
triggers = self.contract_workflow["on"]
self.assertEqual(set(triggers), {"pull_request", "push"})
self.assertEqual(triggers["pull_request"], {"paths": CONTRACT_PATHS})
self.assertEqual(
triggers["push"],
{
"branches": ["main"],
"paths": CONTRACT_PATHS,
},
)

steps = self.contract_workflow["jobs"]["contract"]["steps"]
self.assertEqual(
steps[2],
{
"name": "Install contract tools",
"run": (
"python -m pip install --disable-pip-version-check "
"PyYAML==6.0.2 ruff==0.11.13"
),
},
)
self.assertEqual(
steps[3],
{
"name": "Check contract lint and format",
"run": (
"ruff check .github/utils/test_release_image_cache_sync_contract.py\n"
"ruff format --check "
".github/utils/test_release_image_cache_sync_contract.py\n"
),
},
)
self.assertEqual(
steps[4],
{
"name": "Run reusable workflow contract",
"run": "python .github/utils/test_release_image_cache_sync_contract.py",
},
)

def test_reusable_and_job_outputs_are_additive_and_exact(self) -> None:
workflow_outputs = self.workflow["on"]["workflow_call"]["outputs"]
self.assertEqual(
workflow_outputs["image-index-digest"]["value"],
"${{ jobs.release-image.outputs.image-index-digest }}",
)

job_outputs = self.workflow["jobs"]["release-image"]["outputs"]
self.assertEqual(
job_outputs["tag-name"], "${{ steps.get_tag_name.outputs.TAG-NAME }}"
)
self.assertEqual(
job_outputs["image-index-digest"],
"${{ steps.select_image_index_digest.outputs.image-index-digest }}",
)

def test_both_build_branches_export_direct_digest_and_disable_attestations(
self,
) -> None:
steps = self.workflow["jobs"]["release-image"]["steps"]
token_build = step_by_id(self.workflow, "build_with_args_token")
plain_build = step_by_id(self.workflow, "build_without_args_token")
build_steps = [
step for step in steps if step.get("uses") == "docker/build-push-action@v5"
]

self.assertEqual(build_steps, [token_build, plain_build])
self.assertLess(steps.index(token_build), steps.index(self.selector))
self.assertLess(steps.index(plain_build), steps.index(self.selector))
self.assertEqual(
token_build["if"],
"${{ env.DOCKER_USER != '' && env.DOCKER_PASSWORD != '' && inputs.ARGS_TOKEN }}",
)
self.assertEqual(
plain_build["if"],
"${{ env.DOCKER_USER != '' && env.DOCKER_PASSWORD != '' && ! inputs.ARGS_TOKEN }}",
)
for build in (token_build, plain_build):
self.assertEqual(build["uses"], "docker/build-push-action@v5")
self.assertIs(build["with"]["push"], True)
self.assertIs(build["with"]["provenance"], False)
self.assertIs(build["with"]["sbom"], False)

self.assertEqual(
self.selector["env"],
{
"DIGEST_WITH_ARGS_TOKEN": "${{ steps.build_with_args_token.outputs.digest }}",
"DIGEST_WITHOUT_ARGS_TOKEN": "${{ steps.build_without_args_token.outputs.digest }}",
},
)

def run_selector(
self, token_digest: str = "", plain_digest: str = ""
) -> subprocess.CompletedProcess:
with tempfile.TemporaryDirectory() as tmp:
output = Path(tmp) / "github-output"
env = os.environ.copy()
env.update(
{
"DIGEST_WITH_ARGS_TOKEN": token_digest,
"DIGEST_WITHOUT_ARGS_TOKEN": plain_digest,
"GITHUB_OUTPUT": str(output),
}
)
result = subprocess.run(
["bash", "-c", self.selector["run"]],
check=False,
capture_output=True,
text=True,
env=env,
)
result.github_output = output.read_text() if output.exists() else ""
return result

def test_token_build_digest_is_selected(self) -> None:
result = self.run_selector(token_digest=VALID_DIGEST_A)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(result.github_output, f"image-index-digest={VALID_DIGEST_A}\n")

def test_plain_build_digest_is_selected(self) -> None:
result = self.run_selector(plain_digest=VALID_DIGEST_B)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(result.github_output, f"image-index-digest={VALID_DIGEST_B}\n")

def test_missing_duplicate_and_malformed_digests_fail_closed(self) -> None:
cases = (
("", ""),
(VALID_DIGEST_A, VALID_DIGEST_B),
("sha256:" + "A" * 64, ""),
("sha256:1234", ""),
("not-a-digest", ""),
)
for token_digest, plain_digest in cases:
with self.subTest(token_digest=token_digest, plain_digest=plain_digest):
result = self.run_selector(token_digest, plain_digest)
self.assertNotEqual(result.returncode, 0)
self.assertEqual(result.github_output, "")

def test_selector_has_no_mutable_tag_lookup_fallback(self) -> None:
script = self.selector["run"]
forbidden = (
"docker manifest",
"imagetools",
"skopeo",
"crane",
"TAG-NAME",
"tag-name",
)
self.assertFalse([needle for needle in forbidden if needle in script])


if __name__ == "__main__":
unittest.main()
37 changes: 37 additions & 0 deletions .github/workflows/release-image-cache-sync.yml
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,10 @@ name: Releae Image Cache And Sync To Aliyun

on:
workflow_call:
outputs:
image-index-digest:
description: "OCI index digest produced by the successful image build"
value: ${{ jobs.release-image.outputs.image-index-digest }}
inputs:
MAKE_OPS_PRE:
description: "The pre ops name of makefile (e.g. generate)"
Expand Down Expand Up @@ -114,6 +118,7 @@ jobs:
runs-on: ubuntu-latest
outputs:
tag-name: ${{ steps.get_tag_name.outputs.TAG-NAME }}
image-index-digest: ${{ steps.select_image_index_digest.outputs.image-index-digest }}
steps:
- name: Pre Check
id: pre_check
Expand Down Expand Up @@ -210,6 +215,7 @@ jobs:
password: ${{ env.DOCKER_PASSWORD }}

- if: ${{ env.DOCKER_USER != '' && env.DOCKER_PASSWORD != '' && inputs.ARGS_TOKEN }}
id: build_with_args_token
name: Build and Push to Docker Hub
uses: docker/build-push-action@v5
with:
Expand All @@ -230,6 +236,7 @@ jobs:
sbom: false

- if: ${{ env.DOCKER_USER != '' && env.DOCKER_PASSWORD != '' && ! inputs.ARGS_TOKEN }}
id: build_without_args_token
name: Build and Push to Docker Hub
uses: docker/build-push-action@v5
with:
Expand All @@ -248,6 +255,36 @@ jobs:
provenance: false
sbom: false

- name: Validate and export OCI index digest
id: select_image_index_digest
shell: bash
env:
DIGEST_WITH_ARGS_TOKEN: ${{ steps.build_with_args_token.outputs.digest }}
DIGEST_WITHOUT_ARGS_TOKEN: ${{ steps.build_without_args_token.outputs.digest }}
run: |
set -euo pipefail

digests=()
if [[ -n "$DIGEST_WITH_ARGS_TOKEN" ]]; then
digests+=("$DIGEST_WITH_ARGS_TOKEN")
fi
if [[ -n "$DIGEST_WITHOUT_ARGS_TOKEN" ]]; then
digests+=("$DIGEST_WITHOUT_ARGS_TOKEN")
fi

if [[ ${#digests[@]} -ne 1 ]]; then
echo "expected exactly one successful image build digest, got ${#digests[@]}" >&2
exit 1
fi

image_index_digest="${digests[0]}"
if [[ ! "$image_index_digest" =~ ^sha256:[0-9a-f]{64}$ ]]; then
echo "image build returned an invalid OCI index digest" >&2
exit 1
fi

echo "image-index-digest=$image_index_digest" >> "$GITHUB_OUTPUT"

sync-to-aliyun:
runs-on: ubuntu-latest
if: ${{ inputs.SYNC_ENABLE }}
Expand Down
39 changes: 39 additions & 0 deletions .github/workflows/test-release-image-cache-sync-contract.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
name: Test Release Image Cache Sync Contract

on:
pull_request:
paths:
- '.github/workflows/release-image-cache-sync.yml'
- '.github/workflows/test-release-image-cache-sync-contract.yml'
- '.github/utils/test_release_image_cache_sync_contract.py'
push:
branches:
- main
paths:
- '.github/workflows/release-image-cache-sync.yml'
- '.github/workflows/test-release-image-cache-sync-contract.yml'
- '.github/utils/test_release_image_cache_sync_contract.py'

permissions:
contents: read

jobs:
contract:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4

- uses: actions/setup-python@v5
with:
python-version: '3.12'

- name: Install contract tools
run: python -m pip install --disable-pip-version-check PyYAML==6.0.2 ruff==0.11.13

- name: Check contract lint and format
run: |
ruff check .github/utils/test_release_image_cache_sync_contract.py
ruff format --check .github/utils/test_release_image_cache_sync_contract.py

- name: Run reusable workflow contract
run: python .github/utils/test_release_image_cache_sync_contract.py
Loading