scripts/ci/publication.py
Rama main · réplica de lectura del repositorio original.
'''Validate a guarded publication resume against the original release run.'''
from __future__ import annotations
import argparse
import json
import os
import subprocess
import sys
from collections.abc import Callable, Mapping
from typing import Any
REPOSITORY = 'rhoopr/kei'
RELEASE_WORKFLOW = '.github/workflows/release.yml'
VALIDATE_JOB = 'validate'
QUALIFICATION_JOB = 'qualification / Heavy qualification result'
RELEASE_JOB = 'release'
MAX_TAG_DEPTH = 8
MAX_JOB_PAGES = 100
Api = Callable[[str], Any]
def _mapping(value: Any, what: str) -> Mapping[str, Any]:
if not isinstance(value, Mapping):
raise TypeError('invalid ' + what)
return value
def _sha(value: Any) -> str:
if (
not isinstance(value, str)
or len(value) != 40
or not value.isascii()
or any(character not in '0123456789abcdef' for character in value)
):
raise ValueError('invalid commit SHA')
return value
def _numeric(value: Any) -> bool:
if not isinstance(value, str):
return False
return (
bool(value)
and value.isascii()
and value.isdigit()
and (value == '0' or value[0] != '0')
)
def require_repository(repo: Any) -> str:
if repo != REPOSITORY:
raise ValueError('repository must be rhoopr/kei')
return repo
def require_run(run_id: Any) -> str:
if not _numeric(run_id) or run_id == '0':
raise ValueError('run must be a positive decimal')
return run_id
def require_tag(tag: Any) -> str:
if not isinstance(tag, str) or not tag.startswith('v'):
raise ValueError('tag must be vX.Y.Z')
parts = tag[1:].split('.')
if len(parts) != 3 or not all(_numeric(part) for part in parts):
raise ValueError('tag must be vX.Y.Z')
return tag
def require_release(api: Api, repo: str, tag: str) -> None:
release = _mapping(
api('repos/' + repo + '/releases/tags/' + tag),
'release',
)
if release.get('tag_name') != tag or release.get('draft') is not False:
raise ValueError('matching non-draft release is required')
def resolve_tag(api: Api, repo: str, tag: str) -> str:
ref = _mapping(
api('repos/' + repo + '/git/ref/tags/' + tag),
'tag ref',
)
if ref.get('ref') not in (None, 'refs/tags/' + tag):
raise ValueError('tag ref does not match requested tag')
target = _mapping(ref.get('object'), 'tag target')
seen: set[str] = set()
for _ in range(MAX_TAG_DEPTH):
sha = _sha(target.get('sha'))
if sha in seen:
raise ValueError('tag chain contains a cycle')
seen.add(sha)
kind = target.get('type')
if kind == 'commit':
return sha
if kind != 'tag':
raise ValueError('tag target is not a commit')
annotated = _mapping(
api('repos/' + repo + '/git/tags/' + sha),
'annotated tag',
)
target = _mapping(annotated.get('object'), 'annotated tag target')
raise ValueError('tag chain is too deep')
def _fetch_jobs(api: Api, repo: str, run_id: str) -> list[Mapping[str, Any]]:
collected: list[Mapping[str, Any]] = []
for page in range(1, MAX_JOB_PAGES + 1):
payload = _mapping(
api(
'repos/' + repo + '/actions/runs/' + run_id
+ '/jobs?per_page=100&page=' + str(page)
),
'jobs page',
)
batch = payload.get('jobs')
if not isinstance(batch, list) or not all(isinstance(job, Mapping) for job in batch):
raise ValueError('invalid jobs page')
total = payload.get('total_count')
if not isinstance(total, int) or isinstance(total, bool) or total < 0:
raise ValueError('invalid job count')
collected.extend(batch)
if len(collected) == total:
return collected
if len(collected) > total or not batch:
raise ValueError('job pagination mismatch')
raise ValueError('too many job pages')
def require_jobs(jobs: list[Mapping[str, Any]]) -> None:
for name in (VALIDATE_JOB, QUALIFICATION_JOB, RELEASE_JOB):
matches = [job for job in jobs if job.get('name') == name]
if len(matches) != 1:
raise ValueError('missing or ambiguous required job: ' + name)
job = matches[0]
if job.get('status') != 'completed' or job.get('conclusion') != 'success':
raise ValueError('required job did not succeed: ' + name)
def validate(repo: str, run_id: str, tag: str, api: Api) -> dict[str, str]:
repo = require_repository(repo)
run_id = require_run(run_id)
tag = require_tag(tag)
require_release(api, repo, tag)
latest = _mapping(api('repos/' + repo + '/releases/latest'), 'latest release')
if latest.get('tag_name') != tag:
raise ValueError('resume must target the latest stable release')
sha = resolve_tag(api, repo, tag)
run = _mapping(
api('repos/' + repo + '/actions/runs/' + run_id),
'workflow run',
)
if run.get('status') != 'completed':
raise ValueError('original run must be completed')
if run.get('conclusion') not in ('success', 'failure'):
raise ValueError('original run did not finish with a usable conclusion')
if run.get('event') != 'push':
raise ValueError('original run must be a push')
if run.get('path') != RELEASE_WORKFLOW:
raise ValueError('original run must use the release workflow')
if run.get('head_branch') != tag:
raise ValueError('original run tag does not match')
if run.get('head_sha') != sha:
raise ValueError('original run commit does not match tag')
repository = _mapping(run.get('repository'), 'run repository')
if repository.get('full_name') != repo:
raise ValueError('original run repository does not match')
require_jobs(_fetch_jobs(api, repo, run_id))
return {'sha': sha, 'tag': tag, 'version': tag[1:], 'run_id': run_id}
def write_output(result: Mapping[str, Any]) -> None:
target = os.environ.get('GITHUB_OUTPUT')
if not target:
raise ValueError('GITHUB_OUTPUT is required')
sha = _sha(result.get('sha'))
tag = require_tag(result.get('tag'))
run_id = require_run(result.get('run_id'))
version = result.get('version')
if version != tag[1:]:
raise ValueError('invalid output version')
values = {'sha': sha, 'tag': tag, 'version': version, 'run_id': run_id}
with open(target, 'a', encoding='utf-8') as stream:
for key in ('sha', 'tag', 'version', 'run_id'):
print(key + '=' + values[key], file=stream)
def gh_api(endpoint: str) -> Any:
completed = subprocess.run(
['gh', 'api', endpoint],
check=True,
capture_output=True,
text=True,
)
return json.loads(completed.stdout)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description='Validate a release publication resume')
parser.add_argument('--repo', required=True)
parser.add_argument('--run', required=True, dest='run_id')
parser.add_argument('--tag', required=True)
arguments = parser.parse_args(argv)
try:
result = validate(
arguments.repo,
arguments.run_id,
arguments.tag,
gh_api,
)
write_output(result)
except (TypeError, ValueError, OSError, subprocess.CalledProcessError) as error:
print(error, file=sys.stderr)
return 1
return 0
if __name__ == '__main__':
raise SystemExit(main())