376 lines
11 KiB
Python
376 lines
11 KiB
Python
"""Semantic release versions and channel-aware update selection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from dataclasses import dataclass
|
|
import json
|
|
import re
|
|
from typing import Iterable, Sequence
|
|
|
|
|
|
STABLE_CHANNEL = "stable"
|
|
PRERELEASE_CHANNEL = "prerelease"
|
|
SUPPORTED_CHANNELS = frozenset({STABLE_CHANNEL, PRERELEASE_CHANNEL})
|
|
ACTIVE_STATUS = "active"
|
|
WITHDRAWN_STATUS = "withdrawn"
|
|
BLOCKED_STATUS = "blocked"
|
|
SUPPORTED_RELEASE_STATUSES = frozenset(
|
|
{ACTIVE_STATUS, WITHDRAWN_STATUS, BLOCKED_STATUS}
|
|
)
|
|
|
|
_VERSION_PATTERN = re.compile(
|
|
r"(?P<major>0|[1-9][0-9]*)\."
|
|
r"(?P<minor>0|[1-9][0-9]*)\."
|
|
r"(?P<patch>0|[1-9][0-9]*)"
|
|
r"(?:-(?P<label>beta|rc)\.(?P<number>0|[1-9][0-9]*))?\Z"
|
|
)
|
|
|
|
|
|
class ReleaseModelError(ValueError):
|
|
"""Raised when release metadata violates the supported release model."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SemanticVersion:
|
|
"""The stable, beta, and release-candidate version subset used by releases."""
|
|
|
|
major: int
|
|
minor: int
|
|
patch: int
|
|
prerelease_label: str | None = None
|
|
prerelease_number: int | None = None
|
|
|
|
@classmethod
|
|
def parse(cls, value: str) -> SemanticVersion:
|
|
if not isinstance(value, str):
|
|
raise ReleaseModelError("version must be a string")
|
|
match = _VERSION_PATTERN.fullmatch(value)
|
|
if match is None:
|
|
raise ReleaseModelError(
|
|
"version must use MAJOR.MINOR.PATCH, "
|
|
"MAJOR.MINOR.PATCH-beta.N, or MAJOR.MINOR.PATCH-rc.N"
|
|
)
|
|
label = match.group("label")
|
|
number_text = match.group("number")
|
|
return cls(
|
|
major=int(match.group("major")),
|
|
minor=int(match.group("minor")),
|
|
patch=int(match.group("patch")),
|
|
prerelease_label=label,
|
|
prerelease_number=int(number_text) if number_text is not None else None,
|
|
)
|
|
|
|
@classmethod
|
|
def from_tag(cls, tag: str) -> SemanticVersion:
|
|
if not isinstance(tag, str) or not tag.startswith("v"):
|
|
raise ReleaseModelError("release tag must start with 'v'")
|
|
return cls.parse(tag[1:])
|
|
|
|
@property
|
|
def is_prerelease(self) -> bool:
|
|
return self.prerelease_label is not None
|
|
|
|
@property
|
|
def channel(self) -> str:
|
|
return PRERELEASE_CHANNEL if self.is_prerelease else STABLE_CHANNEL
|
|
|
|
@property
|
|
def tag(self) -> str:
|
|
return f"v{self}"
|
|
|
|
@property
|
|
def target(self) -> tuple[int, int, int]:
|
|
return self.major, self.minor, self.patch
|
|
|
|
def _ordering_key(self) -> tuple[int, int, int, int, int, int]:
|
|
if self.prerelease_label is None:
|
|
return self.major, self.minor, self.patch, 1, 0, 0
|
|
label_rank = 0 if self.prerelease_label == "beta" else 1
|
|
assert self.prerelease_number is not None
|
|
return (
|
|
self.major,
|
|
self.minor,
|
|
self.patch,
|
|
0,
|
|
label_rank,
|
|
self.prerelease_number,
|
|
)
|
|
|
|
def __lt__(self, other: object) -> bool:
|
|
if not isinstance(other, SemanticVersion):
|
|
return NotImplemented
|
|
return self._ordering_key() < other._ordering_key()
|
|
|
|
def __le__(self, other: object) -> bool:
|
|
if not isinstance(other, SemanticVersion):
|
|
return NotImplemented
|
|
return self._ordering_key() <= other._ordering_key()
|
|
|
|
def __gt__(self, other: object) -> bool:
|
|
if not isinstance(other, SemanticVersion):
|
|
return NotImplemented
|
|
return self._ordering_key() > other._ordering_key()
|
|
|
|
def __ge__(self, other: object) -> bool:
|
|
if not isinstance(other, SemanticVersion):
|
|
return NotImplemented
|
|
return self._ordering_key() >= other._ordering_key()
|
|
|
|
def __str__(self) -> str:
|
|
value = f"{self.major}.{self.minor}.{self.patch}"
|
|
if self.prerelease_label is not None:
|
|
value += f"-{self.prerelease_label}.{self.prerelease_number}"
|
|
return value
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReleaseRecord:
|
|
"""The resolver-facing fields shared by channel indexes and release history."""
|
|
|
|
version: str
|
|
channel: str
|
|
status: str = ACTIVE_STATUS
|
|
tag: str | None = None
|
|
manifest_url: str | None = None
|
|
manifest_signature_url: str | None = None
|
|
release_date: str | None = None
|
|
release_notes_url: str | None = None
|
|
compatibility: object | None = None
|
|
|
|
@property
|
|
def semantic_version(self) -> SemanticVersion:
|
|
return SemanticVersion.parse(self.version)
|
|
|
|
def validate(self) -> ReleaseRecord:
|
|
parsed = self.semantic_version
|
|
validate_channel(self.channel)
|
|
if parsed.channel != self.channel:
|
|
raise ReleaseModelError(
|
|
f"version {self.version} belongs to {parsed.channel}, not {self.channel}"
|
|
)
|
|
if self.status not in SUPPORTED_RELEASE_STATUSES:
|
|
raise ReleaseModelError(f"unsupported release status: {self.status}")
|
|
if self.tag is not None and self.tag != parsed.tag:
|
|
raise ReleaseModelError(
|
|
f"release tag {self.tag!r} does not match version {self.version}"
|
|
)
|
|
return self
|
|
|
|
|
|
class UpdateResolver:
|
|
"""Centralize release eligibility for updates, switching, and rollback."""
|
|
|
|
def __init__(self, releases: Iterable[ReleaseRecord]) -> None:
|
|
validated: dict[SemanticVersion, ReleaseRecord] = {}
|
|
for release in releases:
|
|
release.validate()
|
|
parsed = release.semantic_version
|
|
if parsed in validated:
|
|
raise ReleaseModelError(f"duplicate release version: {release.version}")
|
|
validated[parsed] = release
|
|
self._releases = validated
|
|
|
|
def eligible_releases(
|
|
self,
|
|
selected_channel: str,
|
|
*,
|
|
include_inactive: bool = False,
|
|
) -> tuple[ReleaseRecord, ...]:
|
|
validate_channel(selected_channel)
|
|
eligible = []
|
|
for parsed, release in self._releases.items():
|
|
if not include_inactive and release.status != ACTIVE_STATUS:
|
|
continue
|
|
if selected_channel == STABLE_CHANNEL and parsed.is_prerelease:
|
|
continue
|
|
eligible.append(release)
|
|
eligible.sort(key=lambda item: item.semantic_version, reverse=True)
|
|
return tuple(eligible)
|
|
|
|
def latest_eligible(self, selected_channel: str) -> ReleaseRecord | None:
|
|
eligible = self.eligible_releases(selected_channel)
|
|
return eligible[0] if eligible else None
|
|
|
|
def update_candidate(
|
|
self,
|
|
installed_version: str,
|
|
selected_channel: str,
|
|
) -> ReleaseRecord | None:
|
|
installed = SemanticVersion.parse(installed_version)
|
|
latest = self.latest_eligible(selected_channel)
|
|
if latest is None or latest.semantic_version <= installed:
|
|
return None
|
|
return latest
|
|
|
|
def rollback_candidates(
|
|
self,
|
|
installed_version: str,
|
|
selected_channel: str,
|
|
*,
|
|
show_prereleases: bool | None = None,
|
|
include_inactive: bool = False,
|
|
) -> tuple[ReleaseRecord, ...]:
|
|
installed = SemanticVersion.parse(installed_version)
|
|
validate_channel(selected_channel)
|
|
include_prereleases = (
|
|
selected_channel == PRERELEASE_CHANNEL
|
|
if show_prereleases is None
|
|
else show_prereleases
|
|
)
|
|
candidates = []
|
|
for release in self._releases.values():
|
|
parsed = release.semantic_version
|
|
if not include_inactive and release.status != ACTIVE_STATUS:
|
|
continue
|
|
if parsed >= installed:
|
|
continue
|
|
if parsed.is_prerelease and not include_prereleases:
|
|
continue
|
|
candidates.append(release)
|
|
candidates.sort(key=lambda item: item.semantic_version, reverse=True)
|
|
return tuple(candidates)
|
|
|
|
|
|
def validate_channel(channel: str) -> str:
|
|
if channel not in SUPPORTED_CHANNELS:
|
|
raise ReleaseModelError(
|
|
f"channel must be one of: {', '.join(sorted(SUPPORTED_CHANNELS))}"
|
|
)
|
|
return channel
|
|
|
|
|
|
def validate_channel_version(channel: str, version: str) -> SemanticVersion:
|
|
validate_channel(channel)
|
|
parsed = SemanticVersion.parse(version)
|
|
if parsed.channel != channel:
|
|
raise ReleaseModelError(
|
|
f"{channel} channel cannot publish version {version}; "
|
|
f"the version belongs to {parsed.channel}"
|
|
)
|
|
return parsed
|
|
|
|
|
|
def validate_release_progression(
|
|
channel: str,
|
|
version: str,
|
|
tags: Iterable[str],
|
|
) -> SemanticVersion:
|
|
"""Reject a release that would move its channel index backwards."""
|
|
|
|
target = validate_channel_version(channel, version)
|
|
relevant: list[SemanticVersion] = []
|
|
for tag in tags:
|
|
try:
|
|
candidate = SemanticVersion.from_tag(tag)
|
|
except ReleaseModelError:
|
|
continue
|
|
if channel == STABLE_CHANNEL and candidate.is_prerelease:
|
|
continue
|
|
relevant.append(candidate)
|
|
if relevant:
|
|
latest = max(relevant)
|
|
if target <= latest:
|
|
raise ReleaseModelError(
|
|
f"{channel} release {version} must be newer than existing "
|
|
f"release {latest}"
|
|
)
|
|
return target
|
|
|
|
|
|
def previous_release_tag(
|
|
channel: str,
|
|
version: str,
|
|
tags: Iterable[str],
|
|
) -> str | None:
|
|
"""Select the changelog base required by the two release channels."""
|
|
|
|
target = validate_channel_version(channel, version)
|
|
parsed_tags: list[SemanticVersion] = []
|
|
for tag in tags:
|
|
try:
|
|
candidate = SemanticVersion.from_tag(tag)
|
|
except ReleaseModelError:
|
|
continue
|
|
if candidate < target:
|
|
parsed_tags.append(candidate)
|
|
if channel == STABLE_CHANNEL:
|
|
stable = [candidate for candidate in parsed_tags if not candidate.is_prerelease]
|
|
return max(stable).tag if stable else None
|
|
matching_prereleases = [
|
|
candidate
|
|
for candidate in parsed_tags
|
|
if candidate.is_prerelease and candidate.target == target.target
|
|
]
|
|
if matching_prereleases:
|
|
return max(matching_prereleases).tag
|
|
stable = [candidate for candidate in parsed_tags if not candidate.is_prerelease]
|
|
return max(stable).tag if stable else None
|
|
|
|
|
|
def release_metadata_from_tag(tag: str) -> dict[str, object]:
|
|
version = SemanticVersion.from_tag(tag)
|
|
return {
|
|
"tag": tag,
|
|
"version": str(version),
|
|
"channel": version.channel,
|
|
"prerelease": version.is_prerelease,
|
|
}
|
|
|
|
|
|
def _build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
validate = subparsers.add_parser("validate", help="validate a channel/version pair")
|
|
validate.add_argument("channel", choices=sorted(SUPPORTED_CHANNELS))
|
|
validate.add_argument("version")
|
|
inspect_tag = subparsers.add_parser("inspect-tag", help="derive release metadata")
|
|
inspect_tag.add_argument("tag")
|
|
previous = subparsers.add_parser("previous-tag", help="select a changelog base")
|
|
previous.add_argument("channel", choices=sorted(SUPPORTED_CHANNELS))
|
|
previous.add_argument("version")
|
|
previous.add_argument("tags", nargs="*")
|
|
progression = subparsers.add_parser(
|
|
"validate-next",
|
|
help="validate that a release advances its channel",
|
|
)
|
|
progression.add_argument("channel", choices=sorted(SUPPORTED_CHANNELS))
|
|
progression.add_argument("version")
|
|
progression.add_argument("tags", nargs="*")
|
|
return parser
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
arguments = _build_parser().parse_args(argv)
|
|
try:
|
|
if arguments.command == "validate":
|
|
validate_channel_version(arguments.channel, arguments.version)
|
|
return 0
|
|
if arguments.command == "inspect-tag":
|
|
print(json.dumps(release_metadata_from_tag(arguments.tag), sort_keys=True))
|
|
return 0
|
|
if arguments.command == "previous-tag":
|
|
selected = previous_release_tag(
|
|
arguments.channel,
|
|
arguments.version,
|
|
arguments.tags,
|
|
)
|
|
if selected is not None:
|
|
print(selected)
|
|
return 0
|
|
if arguments.command == "validate-next":
|
|
validate_release_progression(
|
|
arguments.channel,
|
|
arguments.version,
|
|
arguments.tags,
|
|
)
|
|
return 0
|
|
except ReleaseModelError as exc:
|
|
parser_error = _build_parser()
|
|
parser_error.error(str(exc))
|
|
raise AssertionError(f"unhandled command: {arguments.command}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|