/
/
/
1"""
2Support immutable-safe GitHub release workflows.
3
4The command-line interface emits values to ``$GITHUB_OUTPUT`` when requested so
5the release workflows can keep version and artifact validation in tested Python.
6"""
7
8# ruff: noqa: T201
9
10from __future__ import annotations
11
12import argparse
13import hashlib
14import json
15import re
16import subprocess
17import sys
18from collections.abc import Mapping
19from dataclasses import dataclass
20from datetime import UTC, datetime
21from pathlib import Path
22from typing import Any
23
24CHANNEL_PATTERNS = {
25 "stable": re.compile(r"^(?P<base>\d+\.\d+\.\d+)$"),
26 "beta": re.compile(r"^(?P<base>\d+\.\d+\.\d+)b(?P<number>\d+)$"),
27 "rc": re.compile(r"^(?P<base>\d+\.\d+\.\d+)rc(?P<number>\d+)$"),
28 "nightly": re.compile(r"^(?P<base>\d+\.\d+\.\d+)\.dev(?P<number>\d+)$"),
29}
30CHANNEL_BRANCHES = {
31 "stable": "stable",
32 "rc": "stable",
33 "beta": "dev",
34 "nightly": "dev",
35}
36SHA256_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$")
37OCI_REVISION_ANNOTATION = "org.opencontainers.image.revision"
38OCI_WHEEL_ANNOTATION = "io.music-assistant.wheel.sha256"
39FRONTEND_VERSION_PATTERN = re.compile(r"^(?P<base>\d+(?:\.\d+)+)(?:\.post(?P<post>\d+))?$")
40
41
42class ReleaseWorkflowError(RuntimeError):
43 """Report an unsafe or inconsistent release state."""
44
45
46@dataclass(frozen=True)
47class AutoRelease:
48 """Describe the next automatic release decision."""
49
50 version: str
51 should_release: bool
52 previous_tag: str | None
53 commits_since: int
54
55
56@dataclass(frozen=True)
57class Asset:
58 """Describe one exact release asset."""
59
60 name: str
61 size: int
62 sha256: str
63
64
65class GitRepository:
66 """Read release tags and commit relationships from a Git repository."""
67
68 def __init__(self, path: Path) -> None:
69 """
70 Initialize a repository reader.
71
72 :param path: Path to the Git worktree.
73 """
74 self.path = path
75
76 def resolve_commit(self, ref: str) -> str:
77 """
78 Resolve a Git ref to its commit SHA.
79
80 :param ref: Commit, branch, or tag to resolve.
81 """
82 return self._run("rev-parse", f"{ref}^{{commit}}")
83
84 def tags(self, pattern: re.Pattern[str]) -> list[str]:
85 """
86 Return version-sorted tags matching a release pattern.
87
88 :param pattern: Full tag-name pattern to match.
89 """
90 output = self._run("tag", "--sort=-version:refname")
91 return [tag for tag in output.splitlines() if pattern.fullmatch(tag)]
92
93 def all_tags(self) -> set[str]:
94 """Return all tag names."""
95 return set(self._run("tag").splitlines())
96
97 def commits_since(self, tag: str | None, source_sha: str, *, allow_diverged: bool) -> int:
98 """
99 Count source commits after a previous release tag.
100
101 :param tag: Previous release tag, or ``None`` for the full source history.
102 :param source_sha: Exact release source commit.
103 :param allow_diverged: Use the merge base when the tag is on a diverged branch.
104 """
105 if tag is None:
106 return int(self._run("rev-list", "--count", source_sha))
107
108 result = self._run_result("merge-base", "--is-ancestor", tag, source_sha)
109 if result.returncode == 0:
110 base = tag
111 elif result.returncode == 1 and allow_diverged:
112 base = self._run("merge-base", tag, source_sha)
113 elif result.returncode == 1:
114 msg = f"Previous {tag} is not an ancestor of source commit {source_sha}"
115 raise ReleaseWorkflowError(msg)
116 else:
117 raise ReleaseWorkflowError(result.stderr.strip() or f"Unable to compare {tag}")
118 return int(self._run("rev-list", "--count", f"{base}..{source_sha}"))
119
120 def _run(self, *args: str) -> str:
121 result = self._run_result(*args)
122 if result.returncode:
123 raise ReleaseWorkflowError(result.stderr.strip() or f"git {' '.join(args)} failed")
124 return result.stdout.strip()
125
126 def _run_result(self, *args: str) -> subprocess.CompletedProcess[str]:
127 return subprocess.run( # noqa: S603
128 ["git", "-C", str(self.path), *args], # noqa: S607
129 check=False,
130 capture_output=True,
131 text=True,
132 )
133
134
135def channel_branch(channel: str) -> str:
136 """
137 Return the source branch for a release channel.
138
139 :param channel: Release channel name.
140 """
141 try:
142 return CHANNEL_BRANCHES[channel]
143 except KeyError as err:
144 raise ReleaseWorkflowError(f"Unsupported release channel: {channel}") from err
145
146
147def validate_version(channel: str, version: str) -> None:
148 """
149 Validate a version against its release channel.
150
151 :param channel: Release channel name.
152 :param version: Version to validate.
153 """
154 try:
155 pattern = CHANNEL_PATTERNS[channel]
156 except KeyError as err:
157 raise ReleaseWorkflowError(f"Unsupported release channel: {channel}") from err
158 if not pattern.fullmatch(version):
159 raise ReleaseWorkflowError(f"Version {version} is invalid for the {channel} channel")
160
161
162def determine_auto_release(
163 repository: GitRepository,
164 channel: str,
165 source_sha: str,
166 *,
167 now: datetime | None = None,
168) -> AutoRelease:
169 """
170 Calculate the next channel version from Git tags and commit ancestry.
171
172 :param repository: Source Git repository.
173 :param channel: Release channel name.
174 :param source_sha: Exact release source commit.
175 :param now: Timestamp used for nightly versions.
176 """
177 source_sha = repository.resolve_commit(source_sha)
178 stable_tags = repository.tags(CHANNEL_PATTERNS["stable"])
179 beta_tags = repository.tags(CHANNEL_PATTERNS["beta"])
180 rc_tags = repository.tags(CHANNEL_PATTERNS["rc"])
181 nightly_tags = repository.tags(CHANNEL_PATTERNS["nightly"])
182 all_tags = repository.all_tags()
183
184 latest_stable = _first(stable_tags)
185 latest_beta = _first(beta_tags)
186 latest_rc = _first(rc_tags)
187 latest_nightly = _first(nightly_tags)
188
189 if channel == "stable":
190 previous_tag = latest_stable
191 commits_since = repository.commits_since(previous_tag, source_sha, allow_diverged=True)
192 version = _next_stable(latest_stable)
193 elif channel == "beta":
194 previous_tag = latest_beta
195 commits_since = repository.commits_since(previous_tag, source_sha, allow_diverged=False)
196 version = _next_beta(latest_beta, latest_stable)
197 elif channel == "rc":
198 version = _next_rc(latest_rc, latest_beta)
199 version_match = CHANNEL_PATTERNS["rc"].fullmatch(version)
200 assert version_match is not None
201 rc_base = version_match.group("base")
202 latest_rc_match = CHANNEL_PATTERNS["rc"].fullmatch(latest_rc or "")
203 if latest_rc_match and latest_rc_match.group("base") == rc_base:
204 previous_tag = latest_rc
205 else:
206 previous_tag = latest_beta
207 commits_since = repository.commits_since(previous_tag, source_sha, allow_diverged=False)
208 elif channel == "nightly":
209 previous_tag = latest_nightly
210 commits_since = repository.commits_since(previous_tag, source_sha, allow_diverged=False)
211 version = _next_nightly(
212 latest_nightly,
213 latest_stable,
214 now or datetime.now(UTC),
215 )
216 else:
217 raise ReleaseWorkflowError(f"Unsupported release channel: {channel}")
218
219 version = _ensure_unique(version, channel, all_tags)
220 return AutoRelease(
221 version=version,
222 should_release=channel != "nightly" or commits_since >= 2,
223 previous_tag=previous_tag,
224 commits_since=commits_since,
225 )
226
227
228def previous_release_tag(
229 repository: GitRepository,
230 channel: str,
231 version: str,
232) -> str | None:
233 """
234 Return the previous tag used as the release-notes base.
235
236 :param repository: Source Git repository.
237 :param channel: Release channel name.
238 :param version: Version currently being prepared.
239 """
240 validate_version(channel, version)
241 if channel == "stable":
242 tags = repository.tags(CHANNEL_PATTERNS["stable"])
243 elif channel == "beta":
244 combined_pattern = re.compile(r"^\d+\.\d+\.\d+(?:b|rc)\d+$")
245 tags = repository.tags(combined_pattern)
246 elif channel == "nightly":
247 tags = repository.tags(CHANNEL_PATTERNS["nightly"])
248 else:
249 match = CHANNEL_PATTERNS["rc"].fullmatch(version)
250 assert match is not None
251 rc_number = int(match.group("number"))
252 if rc_number > 1:
253 previous_rc = f"{match.group('base')}rc{rc_number - 1}"
254 if previous_rc in repository.all_tags():
255 return previous_rc
256 tags = repository.tags(CHANNEL_PATTERNS["beta"])
257 return next((tag for tag in tags if tag != version), None)
258
259
260def is_current_release(
261 repository: GitRepository,
262 channel: str,
263 version: str,
264) -> tuple[bool, str | None]:
265 """
266 Check whether a release is the newest version for its rolling channel.
267
268 :param repository: Source Git repository.
269 :param channel: Release channel name.
270 :param version: Release version to compare.
271 """
272 validate_version(channel, version)
273 if channel == "stable":
274 pattern = CHANNEL_PATTERNS["stable"]
275 elif channel == "nightly":
276 pattern = CHANNEL_PATTERNS["nightly"]
277 else:
278 pattern = re.compile(r"^\d+\.\d+\.\d+(?:b|rc)\d+$")
279 tags = repository.tags(pattern)
280 latest_tag = max(tags, key=_release_version_key, default=None)
281 return latest_tag is None or _release_version_key(version) >= _release_version_key(
282 latest_tag
283 ), latest_tag
284
285
286def compare_release_versions(current: str, requested: str) -> str:
287 """
288 Compare a deployed release version with a requested release version.
289
290 :param current: Version currently deployed to an alias.
291 :param requested: Version requested for promotion.
292 """
293 return _version_relation(
294 _release_version_key(current),
295 _release_version_key(requested),
296 )
297
298
299def compare_frontend_versions(current: str, requested: str) -> str:
300 """
301 Compare a deployed frontend version with a requested frontend version.
302
303 :param current: Frontend version currently configured for the channel.
304 :param requested: Frontend version requested for deployment.
305 """
306 current_base, current_post = _frontend_version_parts(current)
307 requested_base, requested_post = _frontend_version_parts(requested)
308 width = max(len(current_base), len(requested_base))
309 current_key = (*current_base, *((0,) * (width - len(current_base))), current_post)
310 requested_key = (*requested_base, *((0,) * (width - len(requested_base))), requested_post)
311 return _version_relation(current_key, requested_key)
312
313
314def select_release(release_pages: object, version: str) -> dict[str, Any] | None:
315 """
316 Return the only release whose tag exactly matches a version.
317
318 :param release_pages: Paginated responses from the GitHub releases API.
319 :param version: Exact release tag to select.
320 """
321 if not isinstance(release_pages, list):
322 raise ReleaseWorkflowError("GitHub releases response does not contain pages")
323
324 matches: list[dict[str, Any]] = []
325 for page in release_pages:
326 if not isinstance(page, list):
327 raise ReleaseWorkflowError("GitHub releases response contains an invalid page")
328 for release in page:
329 if not isinstance(release, dict):
330 raise ReleaseWorkflowError("GitHub releases response contains invalid metadata")
331 if release.get("tag_name") == version:
332 matches.append(release)
333
334 if len(matches) > 1:
335 release_ids = ", ".join(str(release.get("id", "unknown")) for release in matches)
336 raise ReleaseWorkflowError(f"Multiple releases match exact tag {version}: {release_ids}")
337 if not matches:
338 return None
339
340 release = matches[0]
341 release_id = release.get("id")
342 if not isinstance(release_id, int) or isinstance(release_id, bool) or release_id <= 0:
343 raise ReleaseWorkflowError(f"Release {version} does not contain a valid id")
344 return release
345
346
347def inspect_assets(
348 version: str,
349 *,
350 directory: Path | None = None,
351 release_json: Path | None = None,
352) -> tuple[Asset, Asset]:
353 """
354 Validate and describe the exact wheel and source distribution.
355
356 :param version: Release version.
357 :param directory: Optional directory containing downloaded assets.
358 :param release_json: Optional GitHub release API response.
359 """
360 expected_names = _expected_asset_names(version)
361 api_assets: dict[str, Asset] | None = None
362 local_assets: dict[str, Asset] | None = None
363
364 if release_json is not None:
365 release = json.loads(release_json.read_text(encoding="utf-8"))
366 raw_assets = release.get("assets")
367 if not isinstance(raw_assets, list):
368 raise ReleaseWorkflowError("Release response does not contain an asset list")
369 if len(raw_assets) != len(expected_names):
370 raise ReleaseWorkflowError(
371 f"Release must contain exactly these assets: {', '.join(expected_names)}"
372 )
373 api_assets = {}
374 for raw_asset in raw_assets:
375 if not isinstance(raw_asset, dict):
376 raise ReleaseWorkflowError("Release response contains invalid asset metadata")
377 name = raw_asset.get("name")
378 size = raw_asset.get("size")
379 state = raw_asset.get("state")
380 digest = raw_asset.get("digest")
381 if (
382 not isinstance(name, str)
383 or not isinstance(size, int)
384 or state != "uploaded"
385 or not isinstance(digest, str)
386 or not SHA256_PATTERN.fullmatch(digest)
387 or size <= 0
388 ):
389 raise ReleaseWorkflowError(f"Release asset metadata is invalid: {name!r}")
390 if name in api_assets:
391 raise ReleaseWorkflowError(f"Release contains duplicate asset {name}")
392 api_assets[name] = Asset(name, size, digest.removeprefix("sha256:"))
393 if set(api_assets) != set(expected_names):
394 raise ReleaseWorkflowError(
395 f"Release must contain exactly these assets: {', '.join(expected_names)}"
396 )
397
398 if directory is not None:
399 files = sorted(path for path in directory.iterdir() if path.is_file())
400 if {path.name for path in files} != set(expected_names):
401 raise ReleaseWorkflowError(
402 f"Asset directory must contain exactly: {', '.join(expected_names)}"
403 )
404 local_assets = {
405 path.name: Asset(path.name, path.stat().st_size, _sha256(path)) for path in files
406 }
407 if any(asset.size <= 0 for asset in local_assets.values()):
408 raise ReleaseWorkflowError("Release assets must not be empty")
409
410 assets = local_assets or api_assets
411 if assets is None:
412 raise ReleaseWorkflowError("An asset directory or release response is required")
413 if local_assets is not None and api_assets is not None and local_assets != api_assets:
414 raise ReleaseWorkflowError("Downloaded assets do not match GitHub release metadata")
415 return assets[expected_names[0]], assets[expected_names[1]]
416
417
418def verify_oci_manifest(
419 manifest: dict[str, Any],
420 source_sha: str,
421 wheel_sha256: str,
422) -> tuple[str, list[str]]:
423 """
424 Validate the exact multi-platform image manifest and provenance annotations.
425
426 :param manifest: Formatted Buildx manifest JSON.
427 :param source_sha: Exact source commit expected in the image annotations.
428 :param wheel_sha256: Wheel digest expected in the image annotations.
429 """
430 digest = manifest.get("digest")
431 if not isinstance(digest, str) or not SHA256_PATTERN.fullmatch(digest):
432 raise ReleaseWorkflowError("OCI manifest does not contain a valid digest")
433 annotations = manifest.get("annotations")
434 if not isinstance(annotations, dict):
435 raise ReleaseWorkflowError("OCI manifest does not contain provenance annotations")
436 if annotations.get(OCI_REVISION_ANNOTATION) != source_sha:
437 raise ReleaseWorkflowError("OCI manifest source SHA does not match the release")
438 if annotations.get(OCI_WHEEL_ANNOTATION) != wheel_sha256:
439 raise ReleaseWorkflowError("OCI manifest wheel digest does not match the release")
440
441 raw_manifests = manifest.get("manifests")
442 if not isinstance(raw_manifests, list):
443 raise ReleaseWorkflowError("OCI image is not a multi-platform manifest")
444 runtime_manifests: dict[str, str] = {}
445 for item in raw_manifests:
446 if not isinstance(item, dict):
447 raise ReleaseWorkflowError("OCI manifest contains an invalid descriptor")
448 platform = item.get("platform")
449 child_digest = item.get("digest")
450 if not isinstance(platform, dict) or not isinstance(child_digest, str):
451 raise ReleaseWorkflowError("OCI manifest descriptor is incomplete")
452 operating_system = platform.get("os")
453 architecture = platform.get("architecture")
454 if operating_system == "unknown" and architecture == "unknown":
455 continue
456 if not SHA256_PATTERN.fullmatch(child_digest):
457 raise ReleaseWorkflowError("OCI child manifest digest is invalid")
458 platform_name = f"{operating_system}/{architecture}"
459 if platform_name in runtime_manifests:
460 raise ReleaseWorkflowError(f"OCI image contains duplicate platform {platform_name}")
461 runtime_manifests[platform_name] = child_digest
462 expected_platforms = {"linux/amd64", "linux/arm64"}
463 if set(runtime_manifests) != expected_platforms:
464 raise ReleaseWorkflowError("OCI image must contain linux/amd64 and linux/arm64")
465 return digest, [runtime_manifests[platform] for platform in sorted(expected_platforms)]
466
467
468def update_addon_release(
469 config_path: Path,
470 changelog_path: Path,
471 *,
472 version: str,
473 release_date: str,
474 notes: str,
475 retain: int = 3,
476) -> None:
477 """
478 Update an add-on channel to one canonical release entry.
479
480 :param config_path: Add-on ``config.yaml`` path.
481 :param changelog_path: Add-on ``CHANGELOG.md`` path.
482 :param version: Server version and image tag.
483 :param release_date: Canonical release date in ``DD.MM.YYYY`` format.
484 :param notes: Published server release notes.
485 :param retain: Total number of distinct changelog releases to retain.
486 """
487 if retain < 1:
488 raise ReleaseWorkflowError("At least one changelog release must be retained")
489 set_addon_version(config_path, version)
490
491 existing = changelog_path.read_text(encoding="utf-8") if changelog_path.exists() else ""
492 blocks = _changelog_blocks(existing)
493 previous_blocks: list[str] = []
494 seen_versions = {version}
495 for block_version, block in blocks:
496 if block_version in seen_versions:
497 continue
498 seen_versions.add(block_version)
499 previous_blocks.append(block)
500 new_block = f"# [{version}] - {release_date}\n\n{notes.strip()}"
501 content = "\n\n\n".join([new_block, *previous_blocks[: retain - 1]]) + "\n"
502 changelog_path.write_text(content, encoding="utf-8")
503
504
505def set_addon_version(config_path: Path, version: str) -> None:
506 """
507 Point an add-on at a release version, leaving the rest of its config untouched.
508
509 :param config_path: Add-on ``config.yaml`` path.
510 :param version: Server version and image tag.
511 """
512 config = config_path.read_text(encoding="utf-8")
513 # a duplicate key would leave the stale one in effect, so rewrite nothing unless it is unique
514 config, replacements = re.subn(
515 r"^version: .+$",
516 f"version: {version}",
517 config,
518 flags=re.MULTILINE,
519 )
520 if replacements != 1:
521 raise ReleaseWorkflowError(f"Expected one version field in {config_path}")
522 config_path.write_text(config, encoding="utf-8")
523
524
525def _next_stable(latest_stable: str | None) -> str:
526 if latest_stable is None:
527 return "0.1.0"
528 major, minor, patch = _version_tuple(latest_stable)
529 return f"{major}.{minor}.{patch + 1}"
530
531
532def _next_beta(latest_beta: str | None, latest_stable: str | None) -> str:
533 if latest_beta is not None:
534 match = CHANNEL_PATTERNS["beta"].fullmatch(latest_beta)
535 assert match is not None
536 beta_base = _version_tuple(match.group("base"))
537 if latest_stable is not None:
538 stable = _version_tuple(latest_stable)
539 if stable[:2] >= beta_base[:2]:
540 return f"{stable[0]}.{stable[1] + 1}.0b1"
541 return f"{match.group('base')}b{int(match.group('number')) + 1}"
542 if latest_stable is not None:
543 major, minor, _patch = _version_tuple(latest_stable)
544 return f"{major}.{minor + 1}.0b1"
545 return "0.1.0b1"
546
547
548def _next_rc(latest_rc: str | None, latest_beta: str | None) -> str:
549 if latest_beta is None:
550 return "0.1.0rc1"
551 beta_match = CHANNEL_PATTERNS["beta"].fullmatch(latest_beta)
552 assert beta_match is not None
553 beta_base = beta_match.group("base")
554 if latest_rc is not None:
555 rc_match = CHANNEL_PATTERNS["rc"].fullmatch(latest_rc)
556 assert rc_match is not None
557 if rc_match.group("base") == beta_base:
558 return f"{beta_base}rc{int(rc_match.group('number')) + 1}"
559 return f"{beta_base}rc1"
560
561
562def _next_nightly(
563 latest_nightly: str | None,
564 latest_stable: str | None,
565 now: datetime,
566) -> str:
567 if latest_stable is None:
568 base = "0.1.0"
569 else:
570 major, minor, _patch = _version_tuple(latest_stable)
571 base = f"{major}.{minor + 1}.0"
572 serial = int(now.astimezone(UTC).strftime("%Y%m%d%H"))
573 if latest_nightly is not None:
574 match = CHANNEL_PATTERNS["nightly"].fullmatch(latest_nightly)
575 assert match is not None
576 if match.group("base") == base:
577 serial = max(serial, int(match.group("number")) + 1)
578 return f"{base}.dev{serial}"
579
580
581def _ensure_unique(version: str, channel: str, tags: set[str]) -> str:
582 while version in tags:
583 match = CHANNEL_PATTERNS[channel].fullmatch(version)
584 assert match is not None
585 base = match.group("base")
586 if channel == "stable":
587 major, minor, patch = _version_tuple(base)
588 version = f"{major}.{minor}.{patch + 1}"
589 elif channel == "beta":
590 version = f"{base}b{int(match.group('number')) + 1}"
591 elif channel == "rc":
592 version = f"{base}rc{int(match.group('number')) + 1}"
593 else:
594 version = f"{base}.dev{int(match.group('number')) + 1}"
595 return version
596
597
598def _version_tuple(version: str) -> tuple[int, int, int]:
599 match = CHANNEL_PATTERNS["stable"].fullmatch(version)
600 if match is None:
601 raise ReleaseWorkflowError(f"Invalid base version: {version}")
602 return tuple(int(part) for part in version.split(".")) # type: ignore[return-value]
603
604
605def _release_version_key(version: str) -> tuple[int, int, int, int, int]:
606 for stage, channel in enumerate(("nightly", "beta", "rc", "stable")):
607 match = CHANNEL_PATTERNS[channel].fullmatch(version)
608 if match is None:
609 continue
610 major, minor, patch = _version_tuple(match.group("base"))
611 number = int(match.groupdict().get("number") or 0)
612 return major, minor, patch, stage, number
613 raise ReleaseWorkflowError(f"Unsupported release version: {version}")
614
615
616def _frontend_version_parts(version: str) -> tuple[tuple[int, ...], int]:
617 match = FRONTEND_VERSION_PATTERN.fullmatch(version)
618 if match is None:
619 raise ReleaseWorkflowError(f"Unsupported frontend version: {version}")
620 base = tuple(int(part) for part in match.group("base").split("."))
621 post = int(match.group("post")) if match.group("post") is not None else -1
622 return base, post
623
624
625def _version_relation(current: tuple[int, ...], requested: tuple[int, ...]) -> str:
626 if current > requested:
627 return "newer"
628 if current < requested:
629 return "older"
630 return "equal"
631
632
633def _first(values: list[str]) -> str | None:
634 return values[0] if values else None
635
636
637def _expected_asset_names(version: str) -> tuple[str, str]:
638 return (
639 f"music_assistant-{version}-py3-none-any.whl",
640 f"music_assistant-{version}.tar.gz",
641 )
642
643
644def _sha256(path: Path) -> str:
645 digest = hashlib.sha256()
646 with path.open("rb") as file_handle:
647 for chunk in iter(lambda: file_handle.read(1024 * 1024), b""):
648 digest.update(chunk)
649 return digest.hexdigest()
650
651
652def _changelog_blocks(content: str) -> list[tuple[str, str]]:
653 header_pattern = re.compile(r"(?m)^# \[([^\]]+)\] - .+$")
654 matches = list(header_pattern.finditer(content))
655 if not matches:
656 if content.strip():
657 raise ReleaseWorkflowError("Existing changelog has no release headings")
658 return []
659 if content[: matches[0].start()].strip():
660 raise ReleaseWorkflowError("Existing changelog has content before its first release")
661 blocks: list[tuple[str, str]] = []
662 for index, match in enumerate(matches):
663 end = matches[index + 1].start() if index + 1 < len(matches) else len(content)
664 blocks.append((match.group(1), content[match.start() : end].strip()))
665 return blocks
666
667
668def _write_outputs(
669 values: Mapping[str, str | int | bool | None],
670 output_path: Path | None,
671) -> None:
672 lines = [
673 f"{key}={str(value).lower() if isinstance(value, bool) else '' if value is None else value}"
674 for key, value in values.items()
675 ]
676 if output_path is not None:
677 with output_path.open("a", encoding="utf-8") as file_handle:
678 file_handle.write("\n".join(lines) + "\n")
679 print("\n".join(lines))
680
681
682def _asset_outputs(assets: tuple[Asset, Asset]) -> dict[str, str | int]:
683 wheel, source = assets
684 return {
685 "wheel_name": wheel.name,
686 "wheel_size": wheel.size,
687 "wheel_sha256": wheel.sha256,
688 "sdist_name": source.name,
689 "sdist_size": source.size,
690 "sdist_sha256": source.sha256,
691 }
692
693
694def _configure_release_parser(parser: argparse.ArgumentParser) -> None:
695 parser.add_argument("--version", required=True)
696 parser.add_argument("--releases-json", type=Path, required=True)
697 parser.add_argument("--release-json", type=Path, required=True)
698 parser.add_argument("--github-output", type=Path)
699
700
701def _configure_addon_parser(parser: argparse.ArgumentParser) -> None:
702 parser.add_argument("--config", type=Path, required=True)
703 parser.add_argument("--version", required=True)
704
705
706def _build_parser() -> argparse.ArgumentParser:
707 parser = argparse.ArgumentParser(description=__doc__)
708 subparsers = parser.add_subparsers(dest="command", required=True)
709
710 branch_parser = subparsers.add_parser("branch")
711 branch_parser.add_argument("--channel", required=True)
712 branch_parser.add_argument("--github-output", type=Path)
713
714 validate_parser = subparsers.add_parser("validate-version")
715 validate_parser.add_argument("--channel", required=True)
716 validate_parser.add_argument("--version", required=True)
717
718 auto_parser = subparsers.add_parser("auto-version")
719 auto_parser.add_argument("--channel", required=True)
720 auto_parser.add_argument("--source-sha", required=True)
721 auto_parser.add_argument("--repository", type=Path, default=Path.cwd())
722 auto_parser.add_argument("--timestamp")
723 auto_parser.add_argument("--github-output", type=Path)
724
725 previous_parser = subparsers.add_parser("previous-tag")
726 previous_parser.add_argument("--channel", required=True)
727 previous_parser.add_argument("--version", required=True)
728 previous_parser.add_argument("--repository", type=Path, default=Path.cwd())
729 previous_parser.add_argument("--github-output", type=Path)
730
731 current_parser = subparsers.add_parser("current-version")
732 current_parser.add_argument("--channel", required=True)
733 current_parser.add_argument("--version", required=True)
734 current_parser.add_argument("--repository", type=Path, default=Path.cwd())
735 current_parser.add_argument("--github-output", type=Path)
736
737 release_order_parser = subparsers.add_parser("compare-release-versions")
738 release_order_parser.add_argument("--current", required=True)
739 release_order_parser.add_argument("--requested", required=True)
740 release_order_parser.add_argument("--github-output", type=Path)
741
742 frontend_order_parser = subparsers.add_parser("compare-frontend-versions")
743 frontend_order_parser.add_argument("--current", required=True)
744 frontend_order_parser.add_argument("--requested", required=True)
745 frontend_order_parser.add_argument("--github-output", type=Path)
746
747 _configure_release_parser(subparsers.add_parser("select-release"))
748
749 assets_parser = subparsers.add_parser("assets")
750 assets_parser.add_argument("--version", required=True)
751 assets_parser.add_argument("--directory", type=Path)
752 assets_parser.add_argument("--release-json", type=Path)
753 assets_parser.add_argument("--github-output", type=Path)
754
755 manifest_parser = subparsers.add_parser("verify-manifest")
756 manifest_parser.add_argument("--manifest-json", type=Path, required=True)
757 manifest_parser.add_argument("--source-sha", required=True)
758 manifest_parser.add_argument("--wheel-sha256", required=True)
759 manifest_parser.add_argument("--github-output", type=Path)
760
761 addon_parser = subparsers.add_parser("update-addon")
762 _configure_addon_parser(addon_parser)
763 addon_parser.add_argument("--changelog", type=Path, required=True)
764 addon_parser.add_argument("--release-date", required=True)
765 addon_parser.add_argument("--notes", type=Path, required=True)
766 addon_parser.add_argument("--retain", type=int, default=3)
767
768 _configure_addon_parser(subparsers.add_parser("set-addon-version"))
769 return parser
770
771
772def main() -> int:
773 """Run a release workflow helper command."""
774 args = _build_parser().parse_args()
775 try:
776 if args.command == "branch":
777 _write_outputs({"branch": channel_branch(args.channel)}, args.github_output)
778 elif args.command == "validate-version":
779 validate_version(args.channel, args.version)
780 elif args.command == "auto-version":
781 timestamp = datetime.fromisoformat(args.timestamp) if args.timestamp else None
782 decision = determine_auto_release(
783 GitRepository(args.repository),
784 args.channel,
785 args.source_sha,
786 now=timestamp,
787 )
788 _write_outputs(
789 {
790 "version": decision.version,
791 "should_release": decision.should_release,
792 "previous_tag": decision.previous_tag,
793 "commits_since": decision.commits_since,
794 },
795 args.github_output,
796 )
797 elif args.command == "previous-tag":
798 tag = previous_release_tag(
799 GitRepository(args.repository),
800 args.channel,
801 args.version,
802 )
803 _write_outputs({"previous_tag": tag}, args.github_output)
804 elif args.command == "current-version":
805 is_current, latest_tag = is_current_release(
806 GitRepository(args.repository),
807 args.channel,
808 args.version,
809 )
810 _write_outputs(
811 {
812 "is_current": is_current,
813 "latest_channel_tag": latest_tag,
814 },
815 args.github_output,
816 )
817 elif args.command == "compare-release-versions":
818 _write_outputs(
819 {
820 "current_relation": compare_release_versions(
821 args.current,
822 args.requested,
823 )
824 },
825 args.github_output,
826 )
827 elif args.command == "compare-frontend-versions":
828 _write_outputs(
829 {
830 "current_relation": compare_frontend_versions(
831 args.current,
832 args.requested,
833 )
834 },
835 args.github_output,
836 )
837 elif args.command == "select-release":
838 release_pages = json.loads(args.releases_json.read_text(encoding="utf-8"))
839 release = select_release(release_pages, args.version)
840 if release is None:
841 args.release_json.unlink(missing_ok=True)
842 _write_outputs(
843 {"release_exists": False, "release_id": None},
844 args.github_output,
845 )
846 else:
847 args.release_json.write_text(
848 json.dumps(release, indent=2) + "\n",
849 encoding="utf-8",
850 )
851 _write_outputs(
852 {"release_exists": True, "release_id": release["id"]},
853 args.github_output,
854 )
855 elif args.command == "assets":
856 assets = inspect_assets(
857 args.version,
858 directory=args.directory,
859 release_json=args.release_json,
860 )
861 _write_outputs(_asset_outputs(assets), args.github_output)
862 elif args.command == "verify-manifest":
863 manifest = json.loads(args.manifest_json.read_text(encoding="utf-8"))
864 digest, runtime_digests = verify_oci_manifest(
865 manifest,
866 args.source_sha,
867 args.wheel_sha256,
868 )
869 _write_outputs(
870 {
871 "digest": digest,
872 "runtime_digests": " ".join(runtime_digests),
873 },
874 args.github_output,
875 )
876 elif args.command == "update-addon":
877 update_addon_release(
878 args.config,
879 args.changelog,
880 version=args.version,
881 release_date=args.release_date,
882 notes=args.notes.read_text(encoding="utf-8"),
883 retain=args.retain,
884 )
885 elif args.command == "set-addon-version":
886 set_addon_version(args.config, args.version)
887 return 0
888 except (OSError, ReleaseWorkflowError, ValueError, json.JSONDecodeError) as err:
889 print(f"ERROR: {err}", file=sys.stderr)
890 return 1
891
892
893if __name__ == "__main__":
894 sys.exit(main())
895