Public Access
Emil Simeonov
48e5dd5a34
The packages-API manifest.json blob sha256 is not pullable via repo@sha256:... — production deploy 6251 failed pulling the recorded digest (404 not found). Existence check stays on the packages API (keeps the 404 retry), but the digest now comes from the registry v2 Docker-Content-Digest header. Implements DEVX-173 REQ-1/REQ-2.
534 lines
20 KiB
Python
534 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
# Implements: REQ-5
|
|
"""Auto-create an infra PR to bump a pinned dependency version.
|
|
|
|
After grm or sso-bridge publishes a new package version, this module
|
|
creates a PR in the infra repo to bump the pinned version in
|
|
``pyproject.toml`` or ``ansible/group_vars/all/images.yml``.
|
|
|
|
Reuses ``devx.tools.create_pr`` for PR creation and Vikunja task linking.
|
|
|
|
Usage:
|
|
python -m devx.ci.create_dependency_pr \
|
|
--repo oblachno/infra \
|
|
--package grm \
|
|
--new-version 0.5.2 \
|
|
--source-repo oblachno/grm \
|
|
--source-run-id 12345
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import subprocess # nosec B404
|
|
import tempfile
|
|
import time
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
import click
|
|
import requests
|
|
from dotenv import load_dotenv
|
|
|
|
from devx.api_clients import GiteaClient
|
|
from devx.config import GITEA_API_URL, VIKUNJA_API_URL, VIKUNJA_PROJECT_ID
|
|
from devx.exceptions import APIError
|
|
from devx.i18n import _
|
|
from devx.tokens import get_ci_token, get_vikunja_token
|
|
from devx.tools.create_pr import find_existing_pr
|
|
|
|
load_dotenv()
|
|
|
|
# Where infra pins dependency versions
|
|
PYPROJECT_PATH = "pyproject.toml"
|
|
IMAGES_YML_PATH = "ansible/group_vars/all/images.yml"
|
|
ROLE_DEFAULTS_PATH = "ansible/roles/sso_bridge/defaults/main.yml"
|
|
|
|
|
|
def find_pinned_version(package: str, file_path: str) -> str | None:
|
|
"""Find the currently pinned version of a package in a file.
|
|
|
|
Looks for patterns like:
|
|
- ``"grm @ git+...@v0.5.1"``
|
|
- ``grm = "0.5.1"``
|
|
- ``grm_version: "0.5.1"``
|
|
- ``grm_image_version: "0.5.1"``
|
|
"""
|
|
path = Path(file_path)
|
|
if not path.exists():
|
|
return None
|
|
content = path.read_text(encoding="utf-8")
|
|
# Match various pinning patterns
|
|
patterns = [
|
|
rf"{package}\s*@\s*git\+[^@]+@v?([\d.]+)", # pip: package @ git+url@vX.Y.Z
|
|
rf'{package}\s*=\s*"([\d.]+)"', # pyproject: package = "X.Y.Z"
|
|
rf'{package}_version:\s*"([\d.]+)"', # ansible vars: package_version: "X.Y.Z"
|
|
rf'{package}_image_version:\s*"([\d.]+)"', # ansible vars: package_image_version: "X.Y.Z"
|
|
]
|
|
for pat in patterns:
|
|
match = re.search(pat, content)
|
|
if match:
|
|
return match.group(1)
|
|
return None
|
|
|
|
|
|
def update_pinned_version(file_path: str, package: str, old_version: str, new_version: str) -> bool:
|
|
"""Update the pinned version in a file. Returns True if changed."""
|
|
path = Path(file_path)
|
|
if not path.exists():
|
|
return False
|
|
content = path.read_text(encoding="utf-8")
|
|
# Replace old version with new version in package-related lines
|
|
patterns = [
|
|
(rf"({package}\s*@\s*git\+[^@]+@v?){old_version}", rf"\g<1>{new_version}"),
|
|
(rf'({package}\s*=\s*"){old_version}(")', rf"\g<1>{new_version}\g<2>"),
|
|
(rf'({package}_version:\s*"){old_version}(")', rf"\g<1>{new_version}\g<2>"),
|
|
(rf'({package}_image_version:\s*"){old_version}(")', rf"\g<1>{new_version}\g<2>"),
|
|
]
|
|
new_content = content
|
|
changed = False
|
|
for pat, replacement in patterns:
|
|
new_content, n = re.subn(pat, replacement, new_content)
|
|
if n > 0:
|
|
changed = True
|
|
if changed:
|
|
path.write_text(new_content, encoding="utf-8")
|
|
return changed
|
|
|
|
|
|
_MANIFEST_ACCEPT = (
|
|
"application/vnd.oci.image.index.v1+json, "
|
|
"application/vnd.docker.distribution.manifest.list.v2+json, "
|
|
"application/vnd.oci.image.manifest.v1+json, "
|
|
"application/vnd.docker.distribution.manifest.v2+json"
|
|
)
|
|
|
|
|
|
# Implements: REQ-1 — existence check via packages API (keeps the 404-retry
|
|
# semantics for the publish race); the pullable digest is then resolved via
|
|
# the registry v2 API's Docker-Content-Digest header.
|
|
def resolve_container_digest(api_url: str, owner: str, name: str, tag: str, token: str, timeout_s: int = 0) -> str:
|
|
"""Verify the container tag exists, then resolve its pullable OCI digest.
|
|
|
|
The packages API proves the tag was published (and 404s while the
|
|
producer's image-build workflow races us — ``timeout_s`` retries every
|
|
15s). The packages-API ``manifest.json`` blob sha256 is NOT pullable
|
|
via ``repo@sha256:...``, so the digest comes from the registry v2
|
|
``Docker-Content-Digest`` header instead.
|
|
"""
|
|
url = f"{api_url}/packages/{owner}/container/{name}/{tag}/files"
|
|
headers = {"Authorization": f"token {token}"}
|
|
deadline = time.monotonic() + timeout_s
|
|
while True:
|
|
try:
|
|
resp = requests.get(url, headers=headers, timeout=30) # nosec B310
|
|
resp.raise_for_status()
|
|
except requests.HTTPError as e:
|
|
if e.response is not None and e.response.status_code == 404 and time.monotonic() < deadline:
|
|
click.echo(
|
|
_(
|
|
"[dep-pr] {owner}/{name}:{tag} not published yet — retrying.",
|
|
owner=owner,
|
|
name=name,
|
|
tag=tag,
|
|
)
|
|
)
|
|
time.sleep(15)
|
|
continue
|
|
status = e.response.status_code if e.response is not None else "?"
|
|
raise click.ClickException(
|
|
_(
|
|
"Container artifact {owner}/{name}:{tag} not found or unreadable (HTTP {status}). "
|
|
"Refusing to open a dependency PR for an unpublished artifact.",
|
|
owner=owner,
|
|
name=name,
|
|
tag=tag,
|
|
status=status,
|
|
)
|
|
) from e
|
|
except requests.RequestException as e:
|
|
raise click.ClickException(
|
|
_(
|
|
"Registry lookup failed for {owner}/{name}:{tag}: {error}",
|
|
owner=owner,
|
|
name=name,
|
|
tag=tag,
|
|
error=e,
|
|
)
|
|
) from e
|
|
break
|
|
return _resolve_registry_digest(api_url, owner, name, tag, token)
|
|
|
|
|
|
# Implements: REQ-2 — v2 digest resolution fails closed: non-2xx, missing
|
|
# token, or absent Docker-Content-Digest all raise; a blob sha256 is never
|
|
# recorded as a pullable digest.
|
|
def _resolve_registry_digest(api_url: str, owner: str, name: str, tag: str, token: str) -> str:
|
|
registry = api_url.removesuffix("/api/v1").removesuffix("/")
|
|
repo = f"{owner}/{name}"
|
|
try:
|
|
tok_resp = requests.get( # nosec B310
|
|
f"{registry}/v2/token",
|
|
params={"service": "container_registry", "scope": f"repository:{repo}:pull"},
|
|
auth=("ci", token),
|
|
timeout=30,
|
|
)
|
|
tok_resp.raise_for_status()
|
|
bearer = tok_resp.json().get("token", "")
|
|
man_resp = requests.get( # nosec B310
|
|
f"{registry}/v2/{repo}/manifests/{tag}",
|
|
headers={"Authorization": f"Bearer {bearer}", "Accept": _MANIFEST_ACCEPT},
|
|
timeout=30,
|
|
)
|
|
man_resp.raise_for_status()
|
|
except requests.RequestException as e:
|
|
status = getattr(getattr(e, "response", None), "status_code", "?")
|
|
raise click.ClickException(
|
|
_(
|
|
"Registry v2 digest lookup failed for {repo}:{tag} (HTTP {status}): {error}",
|
|
repo=repo,
|
|
tag=tag,
|
|
status=status,
|
|
error=e,
|
|
)
|
|
) from e
|
|
digest = man_resp.headers.get("Docker-Content-Digest", "")
|
|
if not digest:
|
|
raise click.ClickException(
|
|
_("Registry returned no Docker-Content-Digest for {repo}:{tag}.", repo=repo, tag=tag)
|
|
)
|
|
return digest
|
|
|
|
|
|
def update_manifest(file_path: str, section: str, fields: dict[str, str]) -> bool:
|
|
"""Update a release-manifest JSON section in place. Returns True if changed.
|
|
|
|
Implements REQ-1: manifest fields record the immutable release contract
|
|
(version, git ref, image tag, resolved OCI digest) in the target repo.
|
|
"""
|
|
import json as _json
|
|
|
|
path = Path(file_path)
|
|
if not path.exists():
|
|
raise click.ClickException(_("Manifest file not found: {file}", file=file_path))
|
|
try:
|
|
manifest = _json.loads(path.read_text(encoding="utf-8"))
|
|
except _json.JSONDecodeError as e:
|
|
raise click.ClickException(_("Manifest file {file} is not valid JSON: {error}", file=file_path, error=e)) from e
|
|
existing = manifest.get(section)
|
|
if not isinstance(existing, dict):
|
|
raise click.ClickException(
|
|
_("Manifest file {file} has no object section {section}", file=file_path, section=section)
|
|
)
|
|
changed = False
|
|
for k, v in fields.items():
|
|
if existing.get(k) != v:
|
|
existing[k] = v
|
|
changed = True
|
|
if changed:
|
|
path.write_text(_json.dumps(manifest, indent=2) + "\n", encoding="utf-8")
|
|
return changed
|
|
|
|
|
|
def read_manifest_version(file_path: str, section: str) -> str | None:
|
|
"""Read the currently pinned version from a release manifest section."""
|
|
import json as _json
|
|
|
|
path = Path(file_path)
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
manifest = _json.loads(path.read_text(encoding="utf-8"))
|
|
except _json.JSONDecodeError:
|
|
return None
|
|
existing = manifest.get(section)
|
|
if isinstance(existing, dict):
|
|
version = existing.get("version")
|
|
return str(version) if version is not None else None
|
|
return None
|
|
|
|
|
|
def create_vikunja_task(title: str, description: str, project_id: int = 0) -> str | None:
|
|
"""Create a Vikunja task and return its identifier (e.g., OBL-INFRA-531)."""
|
|
try:
|
|
token = get_vikunja_token()
|
|
except click.ClickException:
|
|
return None
|
|
|
|
from devx.api_clients import VikunjaClient
|
|
|
|
client = VikunjaClient(VIKUNJA_API_URL, token)
|
|
task = client.create_task(project_id or VIKUNJA_PROJECT_ID, title=title, description=description)
|
|
return str(task.get("identifier", ""))
|
|
|
|
|
|
@click.command()
|
|
@click.option("--repo", default="oblachno/infra", help=_("Target repo (owner/name) to create PR in"))
|
|
@click.option("--package", required=True, help=_("Package name to bump (e.g., grm, sso-bridge)"))
|
|
@click.option("--new-version", required=True, help=_("New version to pin"))
|
|
@click.option("--source-repo", required=True, help=_("Source repo that published (owner/name)"))
|
|
@click.option("--source-run-id", default="", help=_("CI run ID that triggered the publish"))
|
|
@click.option(
|
|
"--manifest",
|
|
"manifest_path",
|
|
default="",
|
|
help=_(
|
|
"Path to a release-manifest JSON in the target repo. When set, the PR updates "
|
|
"the manifest section named after --package instead of a regex version bump."
|
|
),
|
|
)
|
|
@click.option(
|
|
"--verify-container",
|
|
default="",
|
|
help=_(
|
|
"Container to verify before opening the PR (owner/name). Resolves the OCI "
|
|
"digest of the tag matching --new-version and records it in the manifest."
|
|
),
|
|
)
|
|
@click.option(
|
|
"--source-ref",
|
|
default="",
|
|
help=_("Git ref of the producer release (default: v<new-version>), recorded in the manifest."),
|
|
)
|
|
@click.option(
|
|
"--container-tag",
|
|
default="",
|
|
help=_(
|
|
"Container tag to verify with --verify-container (default: --new-version). "
|
|
"Use when the image tag differs from the release version, e.g. a package "
|
|
"__version__ tag vs a release git tag."
|
|
),
|
|
)
|
|
@click.option(
|
|
"--verify-timeout",
|
|
type=int,
|
|
default=600,
|
|
help=_(
|
|
"Seconds to keep retrying --verify-container while the artifact returns 404 "
|
|
"(the image build races this step). 0 disables retries."
|
|
),
|
|
)
|
|
@click.option(
|
|
"--task-project-id",
|
|
type=int,
|
|
default=0,
|
|
help=_(
|
|
"Vikunja project for the tracking task (default: DEVX_VIKUNJA_PROJECT_ID). "
|
|
"Use the target repo's project so the generated branch/spec satisfy its validation."
|
|
),
|
|
)
|
|
@click.option("--dry-run", is_flag=True, default=False, help=_("Show what would be done without creating PR"))
|
|
def cli(
|
|
repo: str,
|
|
package: str,
|
|
new_version: str,
|
|
source_repo: str,
|
|
source_run_id: str,
|
|
manifest_path: str,
|
|
verify_container: str,
|
|
source_ref: str,
|
|
container_tag: str,
|
|
verify_timeout: int,
|
|
task_project_id: int,
|
|
dry_run: bool,
|
|
) -> None:
|
|
"""Create an infra PR to bump a pinned dependency version."""
|
|
token = get_ci_token()
|
|
if "/" not in repo:
|
|
raise click.ClickException(_("Invalid repo format: {repo}", repo=repo))
|
|
owner, repo_name = repo.split("/", 1)
|
|
client = GiteaClient(GITEA_API_URL, token, owner, repo_name)
|
|
|
|
# Implements: REQ-1 — verify the producer artifact exists and resolve its
|
|
# digest before any branch/PR work begins.
|
|
image_digest = ""
|
|
if verify_container:
|
|
if "/" not in verify_container:
|
|
raise click.ClickException(
|
|
_("Invalid container format: {container} (expected owner/name)", container=verify_container)
|
|
)
|
|
c_owner, c_name = verify_container.split("/", 1)
|
|
image_tag = container_tag or new_version
|
|
image_digest = resolve_container_digest(GITEA_API_URL, c_owner, c_name, image_tag, token, verify_timeout)
|
|
click.echo(
|
|
_(
|
|
"[dep-pr] Verified {container}:{version} -> {digest}",
|
|
container=verify_container,
|
|
version=image_tag,
|
|
digest=image_digest,
|
|
)
|
|
)
|
|
|
|
# Clone the target repo — this tool runs from the *producer* repo's CI,
|
|
# so every file lookup and git operation must happen inside a clone of
|
|
# the target repo, not the producer checkout in CWD.
|
|
workdir = Path(tempfile.mkdtemp(prefix="dep-pr-"))
|
|
clone_url = f"{GITEA_API_URL.removesuffix('/api/v1')}/{repo}.git"
|
|
auth_cfg = f"http.extraHeader=Authorization: token {token}"
|
|
subprocess.run( # nosec B603 B607
|
|
["git", "-c", auth_cfg, "clone", "--depth", "50", clone_url, str(workdir)],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
|
|
# Find current pinned version
|
|
old_version = None
|
|
changed_file = None
|
|
if manifest_path:
|
|
old_version = read_manifest_version(str(workdir / manifest_path), package)
|
|
changed_file = manifest_path if old_version else None
|
|
else:
|
|
for f in [PYPROJECT_PATH, IMAGES_YML_PATH, ROLE_DEFAULTS_PATH]:
|
|
old_version = find_pinned_version(package, str(workdir / f))
|
|
if old_version:
|
|
changed_file = f
|
|
break
|
|
|
|
if not old_version:
|
|
click.echo(_("[dep-pr] Could not find pinned version for {pkg} in infra repo.", pkg=package))
|
|
if dry_run:
|
|
return
|
|
raise click.ClickException(_("Could not find pinned version for {pkg}", pkg=package))
|
|
|
|
if old_version == new_version:
|
|
click.echo(_("[dep-pr] {pkg} already at {version} — no PR needed.", pkg=package, version=new_version))
|
|
return
|
|
|
|
click.echo(
|
|
_(
|
|
"[dep-pr] Bumping {pkg} from {old} to {new} in {file}",
|
|
pkg=package,
|
|
old=old_version,
|
|
new=new_version,
|
|
file=changed_file,
|
|
)
|
|
)
|
|
|
|
if dry_run:
|
|
click.echo(f"[dep-pr] DRY RUN: would update {changed_file} and create PR")
|
|
return
|
|
|
|
# Implements: REQ-1 — create the tracking task in the *target* repo's
|
|
# Vikunja project so its identifier satisfies the target's branch/PR-title
|
|
# validation (e.g., OBL-INFRA-N for oblachno/infra).
|
|
task_title = f"Bump {package} to {new_version}"
|
|
task_desc = (
|
|
f"<p>Auto-created dependency bump PR.</p>"
|
|
f"<p>Package: {package}</p>"
|
|
f"<p>Version: {old_version} → {new_version}</p>"
|
|
f"<p>Source: {source_repo} (run #{source_run_id})</p>"
|
|
)
|
|
task_id = create_vikunja_task(task_title, task_desc, task_project_id)
|
|
|
|
# Create a branch — embed the task ID so target-repo validation accepts it.
|
|
branch_name = f"deps/{task_id}-{package}-{new_version}" if task_id else f"deps/{package}-{new_version}"
|
|
base_branch = "master"
|
|
|
|
# Check for existing PR (reuse from tools.create_pr)
|
|
existing = find_existing_pr(client, branch_name)
|
|
if existing:
|
|
click.echo(_("[dep-pr] PR already exists: #{number}", number=existing.get("number", "?")))
|
|
return
|
|
|
|
# Create branch via API
|
|
try:
|
|
# Implements: REQ-1 — Gitea lacks POST /git/refs; create the branch
|
|
# from master via the branches API.
|
|
client._request(
|
|
"POST",
|
|
"/branches",
|
|
json={"new_branch_name": branch_name, "old_branch_name": base_branch},
|
|
)
|
|
except APIError as e:
|
|
if "already exists" in str(e).lower():
|
|
click.echo(f"[dep-pr] Branch {branch_name} already exists")
|
|
else:
|
|
raise click.ClickException(_("Failed to create branch: {error}", error=str(e))) from None
|
|
|
|
# Check out the API-created branch inside the target clone. A plain
|
|
# fetch only populates FETCH_HEAD — fetch into the remote-tracking ref
|
|
# and force-create the local branch from it.
|
|
subprocess.run( # nosec B603 B607
|
|
["git", "fetch", "origin", f"{branch_name}:refs/remotes/origin/{branch_name}"],
|
|
check=True,
|
|
capture_output=True,
|
|
cwd=workdir,
|
|
)
|
|
subprocess.run( # nosec B603 B607
|
|
["git", "checkout", "-B", branch_name, f"origin/{branch_name}"],
|
|
check=True,
|
|
capture_output=True,
|
|
cwd=workdir,
|
|
)
|
|
|
|
if manifest_path:
|
|
fields = {
|
|
"version": new_version,
|
|
"git_ref": source_ref or f"v{new_version}",
|
|
"image_tag": container_tag or new_version,
|
|
"updated_at": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
}
|
|
if image_digest:
|
|
fields["image_digest"] = image_digest
|
|
if source_run_id:
|
|
fields["source_run_id"] = source_run_id
|
|
if not update_manifest(str(workdir / manifest_path), package, fields):
|
|
raise click.ClickException(_("Failed to update {file}", file=manifest_path))
|
|
elif not changed_file or not update_pinned_version(str(workdir / changed_file), package, old_version, new_version):
|
|
raise click.ClickException(_("Failed to update {file}", file=changed_file))
|
|
|
|
assert changed_file is not None # nosec B101 — narrowed by the early exit above
|
|
add_files = [changed_file]
|
|
if task_id:
|
|
# Implements: REQ-1 — spec-driven validation requires a spec file.
|
|
spec_rel = f"docs/specs/{task_id}.md"
|
|
spec_file = workdir / spec_rel
|
|
spec_file.parent.mkdir(parents=True, exist_ok=True)
|
|
spec_file.write_text(
|
|
f"# {task_id}: {task_title}\n\n"
|
|
f"## Problem\n\n"
|
|
f"{source_repo} released {package} {new_version}; this repo pins {old_version}.\n\n"
|
|
f"## Approach\n\n"
|
|
f"REQ-1: Update `{changed_file}` to pin {package} {new_version} "
|
|
f"(auto-generated dependency PR).\n\n"
|
|
f"## Test Plan\n\n"
|
|
f"- Producer release CI verified the artifact "
|
|
f"({source_repo} run #{source_run_id or 'n/a'}).\n\n"
|
|
f"## Deploy Plan\n\n"
|
|
f"Merge updates the pin; the next deploy applies it.\n\n"
|
|
f"## Rollback Plan\n\n"
|
|
f"Revert the pin bump.\n\n"
|
|
f"## Acceptance Criteria\n\n"
|
|
f"- [x] REQ-1: `{changed_file}` pins {package} {new_version}.\n",
|
|
encoding="utf-8",
|
|
)
|
|
add_files.append(spec_rel)
|
|
subprocess.run(["git", "add", *add_files], check=True, cwd=workdir) # nosec B603 B607
|
|
commit_msg = f"deps: bump {package} from {old_version} to {new_version}"
|
|
subprocess.run(["git", "commit", "-m", commit_msg], check=True, cwd=workdir) # nosec B603 B607
|
|
subprocess.run( # nosec B603 B607
|
|
["git", "-c", auth_cfg, "push", "origin", branch_name],
|
|
check=True,
|
|
cwd=workdir,
|
|
)
|
|
|
|
# Create PR directly (dependency PRs have custom titles, not Vikunja-derived)
|
|
pr_title = f"{task_id}: {task_title}" if task_id else task_title
|
|
pr_body = (
|
|
f"## Dependency Bump\n\n"
|
|
f"Bumps **{package}** from `{old_version}` to `{new_version}`.\n\n"
|
|
f"- **Source**: {source_repo}\n"
|
|
f"- **Triggered by**: CI run #{source_run_id}\n"
|
|
f"- **Changed file**: `{changed_file}`\n\n"
|
|
f"This PR was auto-created by `devx.ci.create_dependency_pr`.\n"
|
|
)
|
|
if task_id:
|
|
pr_body += f"\nCloses {task_id}"
|
|
|
|
pr = client.create_pr(title=pr_title, head=branch_name, base=base_branch, body=pr_body)
|
|
click.echo(_("[dep-pr] Created PR #{number}: {title}", number=pr.get("number", "?"), title=pr_title))
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
cli()
|