feat(pkg-sync): add rpm upstream providers
This commit is contained in:
+348
-154
@@ -8,6 +8,7 @@ import argparse
|
||||
import json
|
||||
import os
|
||||
import platform
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
@@ -34,6 +35,14 @@ ARCH_VERSION_UPDATE_AVAILABLE: Final = 2
|
||||
RPM_QUERY_FORMAT: Final = "%{NAME}\t%{EPOCH}\t%{VERSION}\t%{RELEASE}\t%{ARCH}\n"
|
||||
RPM_QUERY_FIELD_COUNT: Final = 5
|
||||
GITHUB_RELEASE_PATH_PART_COUNT: Final = 5
|
||||
GITEA_ARCHIVE_PATH_PART_COUNT: Final = 4
|
||||
ARCHIVE_SUFFIXES: Final = (
|
||||
".tar.gz",
|
||||
".tar.bz2",
|
||||
".tar.xz",
|
||||
".tar.zst",
|
||||
".zip",
|
||||
)
|
||||
|
||||
TargetT = TypeVar("TargetT")
|
||||
|
||||
@@ -101,12 +110,6 @@ class RpmTarget:
|
||||
return f"{self.package_dir.name}/{self.spec.name}"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RpmVersionLine:
|
||||
index: int
|
||||
value: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class GithubReleaseSource:
|
||||
owner: str
|
||||
@@ -114,6 +117,38 @@ class GithubReleaseSource:
|
||||
tag_template: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class GiteaTagSource:
|
||||
scheme: str
|
||||
host: str
|
||||
owner: str
|
||||
repo: str
|
||||
tag_template: str
|
||||
|
||||
|
||||
ReleaseSource = GithubReleaseSource | GiteaTagSource
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class UpstreamMatch:
|
||||
provider: UpstreamProvider
|
||||
source: ReleaseSource
|
||||
|
||||
|
||||
class UpstreamProvider(ABC):
|
||||
@abstractmethod
|
||||
def match_source(
|
||||
self,
|
||||
source: str,
|
||||
current_version: str,
|
||||
) -> ReleaseSource | None:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def latest_version(self, source: ReleaseSource) -> str | None:
|
||||
pass
|
||||
|
||||
|
||||
def error(message: str) -> None:
|
||||
print(f"error: {message}", file=sys.stderr, flush=True)
|
||||
|
||||
@@ -751,63 +786,41 @@ def rpm_installed_state(metadata: RpmMetadata) -> InstalledState:
|
||||
return compare_expected_versions(rpm_expected_versions(metadata), installed)
|
||||
|
||||
|
||||
def rpm_spec_lines(spec: Path) -> list[str] | None:
|
||||
try:
|
||||
return spec.read_text(encoding="utf-8").splitlines(keepends=True)
|
||||
except OSError as exc:
|
||||
error(f"failed to read {spec}: {exc}")
|
||||
def rpm_spec_version(spec: Path) -> str | None:
|
||||
output = command_output(
|
||||
[
|
||||
"rpmspec",
|
||||
"-q",
|
||||
"--srpm",
|
||||
"--queryformat",
|
||||
"%{VERSION}\n",
|
||||
spec.name,
|
||||
],
|
||||
cwd=spec.parent,
|
||||
)
|
||||
if output is None:
|
||||
return None
|
||||
|
||||
|
||||
def rpm_spec_version_line(spec: Path) -> RpmVersionLine | None:
|
||||
lines = rpm_spec_lines(spec)
|
||||
if lines is None:
|
||||
return None
|
||||
|
||||
for index, line in enumerate(lines):
|
||||
text = line.lstrip()
|
||||
if not text.startswith("Version:"):
|
||||
continue
|
||||
_, _, value = text.partition(":")
|
||||
version = value.strip().split(maxsplit=1)
|
||||
if not version:
|
||||
error(f"empty Version in {spec}")
|
||||
return None
|
||||
return RpmVersionLine(index, version[0])
|
||||
|
||||
error(f"literal Version line not found in {spec}")
|
||||
versions = sorted({line.strip() for line in output.splitlines() if line.strip()})
|
||||
if len(versions) == 1:
|
||||
return versions[0]
|
||||
if versions:
|
||||
error(f"unexpected multiple Version values in {spec}: {', '.join(versions)}")
|
||||
else:
|
||||
error(f"empty Version in {spec}")
|
||||
return None
|
||||
|
||||
|
||||
def rpm_spec_version(spec: Path) -> str | None:
|
||||
version_line = rpm_spec_version_line(spec)
|
||||
if version_line is None:
|
||||
return None
|
||||
return version_line.value
|
||||
def rpm_bump_spec_version(spec: Path, version: str) -> bool:
|
||||
status = command_status(
|
||||
["rpmdev-bumpspec", "-n", version, spec.name],
|
||||
cwd=spec.parent,
|
||||
)
|
||||
if status == EXIT_OK:
|
||||
return True
|
||||
|
||||
|
||||
def rpm_rewrite_spec_version(spec: Path, version: str) -> bool:
|
||||
lines = rpm_spec_lines(spec)
|
||||
version_line = rpm_spec_version_line(spec)
|
||||
if lines is None or version_line is None:
|
||||
return False
|
||||
|
||||
old_line = lines[version_line.index]
|
||||
tag, separator, value = old_line.partition(":")
|
||||
if not separator:
|
||||
error(f"invalid Version line in {spec}")
|
||||
return False
|
||||
|
||||
spacing = value[: len(value) - len(value.lstrip())]
|
||||
newline = "\n" if old_line.endswith("\n") else ""
|
||||
lines[version_line.index] = f"{tag}:{spacing}{version}{newline}"
|
||||
|
||||
try:
|
||||
spec.write_text("".join(lines), encoding="utf-8")
|
||||
except OSError as exc:
|
||||
error(f"failed to update {spec}: {exc}")
|
||||
return False
|
||||
return True
|
||||
error(f"failed to update {spec} to {version}")
|
||||
return False
|
||||
|
||||
|
||||
def rpm_source_name(line: str) -> bool:
|
||||
@@ -831,13 +844,6 @@ def rpm_sources_from_lines(lines: Sequence[str]) -> tuple[str, ...]:
|
||||
return tuple(sources)
|
||||
|
||||
|
||||
def rpm_raw_spec_sources(spec: Path) -> tuple[str, ...] | None:
|
||||
lines = rpm_spec_lines(spec)
|
||||
if lines is None:
|
||||
return None
|
||||
return rpm_sources_from_lines(lines)
|
||||
|
||||
|
||||
def rpm_spec_sources(spec: Path) -> tuple[str, ...] | None:
|
||||
output = command_output(["rpmspec", "--parse", spec.name], cwd=spec.parent)
|
||||
if output is None:
|
||||
@@ -847,7 +853,9 @@ def rpm_spec_sources(spec: Path) -> tuple[str, ...] | None:
|
||||
|
||||
def rpm_source_file_name(source: str) -> str | None:
|
||||
parsed = urllib.parse.urlparse(source)
|
||||
if parsed.scheme:
|
||||
if parsed.scheme and parsed.fragment.startswith("/"):
|
||||
name = Path(urllib.parse.unquote(parsed.fragment)).name
|
||||
elif parsed.scheme:
|
||||
name = Path(urllib.parse.unquote(parsed.path)).name
|
||||
else:
|
||||
name = Path(source).name
|
||||
@@ -858,86 +866,286 @@ def rpm_source_file_name(source: str) -> str | None:
|
||||
return name
|
||||
|
||||
|
||||
def github_release_source(source: str) -> GithubReleaseSource | None:
|
||||
parsed = urllib.parse.urlparse(source)
|
||||
if parsed.scheme not in {"http", "https"}:
|
||||
def release_tag_template(
|
||||
tag: str, current_version: str | None
|
||||
) -> str | None:
|
||||
if tag.count("%{version}") == 1:
|
||||
return tag
|
||||
if current_version is None:
|
||||
return None
|
||||
if parsed.netloc != "github.com":
|
||||
if tag.count(current_version) != 1:
|
||||
return None
|
||||
|
||||
parts = urllib.parse.unquote(parsed.path).strip("/").split("/")
|
||||
if len(parts) < GITHUB_RELEASE_PATH_PART_COUNT:
|
||||
return None
|
||||
if parts[2:4] != ["releases", "download"]:
|
||||
return None
|
||||
|
||||
tag_template = parts[4]
|
||||
if tag_template.count("%{version}") != 1:
|
||||
return None
|
||||
|
||||
return GithubReleaseSource(
|
||||
owner=parts[0],
|
||||
repo=parts[1],
|
||||
tag_template=tag_template,
|
||||
)
|
||||
return tag.replace(current_version, "%{version}", 1)
|
||||
|
||||
|
||||
def github_api_headers() -> dict[str, str]:
|
||||
headers = {
|
||||
"Accept": "application/vnd.github+json",
|
||||
"User-Agent": "pkg-sync",
|
||||
"X-GitHub-Api-Version": "2022-11-28",
|
||||
}
|
||||
token = os.environ.get("GITHUB_TOKEN")
|
||||
if token:
|
||||
headers["Authorization"] = f"Bearer {token}"
|
||||
return headers
|
||||
def archive_tag_name(name: str) -> str | None:
|
||||
for suffix in ARCHIVE_SUFFIXES:
|
||||
if name.endswith(suffix):
|
||||
return name[: -len(suffix)]
|
||||
return None
|
||||
|
||||
|
||||
def github_latest_release_tag(source: GithubReleaseSource) -> str | None:
|
||||
url = (
|
||||
"https://api.github.com/repos/"
|
||||
f"{source.owner}/{source.repo}/releases/latest"
|
||||
)
|
||||
request = urllib.request.Request(url, headers=github_api_headers())
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=60) as response:
|
||||
data = json.load(response)
|
||||
except (OSError, urllib.error.URLError, ValueError) as exc:
|
||||
error(
|
||||
f"failed to check GitHub releases for {source.owner}/{source.repo}: {exc}"
|
||||
)
|
||||
return None
|
||||
|
||||
tag = data.get("tag_name")
|
||||
if not isinstance(tag, str) or not tag:
|
||||
error(
|
||||
f"GitHub latest release for {source.owner}/{source.repo} has no tag"
|
||||
)
|
||||
return None
|
||||
return tag
|
||||
def version_sort_key(version: str) -> tuple[tuple[int, int | str], ...]:
|
||||
parts: list[tuple[int, int | str]] = []
|
||||
for part in re.split(r"([0-9]+)", version):
|
||||
if not part:
|
||||
continue
|
||||
if part.isdigit():
|
||||
parts.append((0, int(part)))
|
||||
else:
|
||||
parts.append((1, part))
|
||||
return tuple(parts)
|
||||
|
||||
|
||||
def github_version_from_tag(
|
||||
release_source: GithubReleaseSource,
|
||||
def version_from_tag(
|
||||
release_source: ReleaseSource,
|
||||
tag: str,
|
||||
*,
|
||||
report_mismatch: bool,
|
||||
) -> str | None:
|
||||
prefix, _, suffix = release_source.tag_template.partition("%{version}")
|
||||
if not tag.startswith(prefix) or not tag.endswith(suffix):
|
||||
error(
|
||||
f"latest GitHub release tag {tag!r} does not match "
|
||||
f"{release_source.tag_template!r}",
|
||||
)
|
||||
if report_mismatch:
|
||||
error(
|
||||
f"latest release tag {tag!r} does not match "
|
||||
f"{release_source.tag_template!r}",
|
||||
)
|
||||
return None
|
||||
|
||||
end = len(tag) - len(suffix) if suffix else len(tag)
|
||||
version = tag[len(prefix) : end]
|
||||
if not version:
|
||||
error(f"empty version in GitHub release tag {tag!r}")
|
||||
error(f"empty version in release tag {tag!r}")
|
||||
return None
|
||||
return version
|
||||
|
||||
|
||||
class GithubProvider(UpstreamProvider):
|
||||
@override
|
||||
def match_source(
|
||||
self,
|
||||
source: str,
|
||||
current_version: str,
|
||||
) -> GithubReleaseSource | None:
|
||||
parsed = urllib.parse.urlparse(source)
|
||||
if parsed.scheme not in {"http", "https"}:
|
||||
return None
|
||||
if parsed.netloc != "github.com":
|
||||
return None
|
||||
|
||||
parts = urllib.parse.unquote(parsed.path).strip("/").split("/")
|
||||
if len(parts) < GITHUB_RELEASE_PATH_PART_COUNT:
|
||||
return None
|
||||
if parts[2:4] != ["releases", "download"]:
|
||||
return None
|
||||
|
||||
tag_template = release_tag_template(parts[4], current_version)
|
||||
if tag_template is None:
|
||||
return None
|
||||
|
||||
return GithubReleaseSource(
|
||||
owner=parts[0],
|
||||
repo=parts[1],
|
||||
tag_template=tag_template,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _api_headers() -> dict[str, str]:
|
||||
headers = {
|
||||
"Accept": "application/vnd.github+json",
|
||||
"User-Agent": "pkg-sync",
|
||||
"X-GitHub-Api-Version": "2022-11-28",
|
||||
}
|
||||
token = os.environ.get("GITHUB_TOKEN")
|
||||
if token:
|
||||
headers["Authorization"] = f"Bearer {token}"
|
||||
return headers
|
||||
|
||||
def _latest_release_tag(self, source: GithubReleaseSource) -> str | None:
|
||||
url = (
|
||||
"https://api.github.com/repos/"
|
||||
f"{source.owner}/{source.repo}/releases/latest"
|
||||
)
|
||||
request = urllib.request.Request(url, headers=self._api_headers())
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=60) as response:
|
||||
data = json.load(response)
|
||||
except (OSError, urllib.error.URLError, ValueError) as exc:
|
||||
error(
|
||||
"failed to check GitHub releases for "
|
||||
f"{source.owner}/{source.repo}: {exc}"
|
||||
)
|
||||
return None
|
||||
|
||||
tag = data.get("tag_name")
|
||||
if not isinstance(tag, str) or not tag:
|
||||
error(
|
||||
"GitHub latest release for "
|
||||
f"{source.owner}/{source.repo} has no tag"
|
||||
)
|
||||
return None
|
||||
return tag
|
||||
|
||||
@override
|
||||
def latest_version(self, source: ReleaseSource) -> str | None:
|
||||
if not isinstance(source, GithubReleaseSource):
|
||||
raise TypeError("GitHub provider received non-GitHub source")
|
||||
|
||||
tag = self._latest_release_tag(source)
|
||||
if tag is None:
|
||||
return None
|
||||
return version_from_tag(source, tag, report_mismatch=True)
|
||||
|
||||
|
||||
class GiteaProvider(UpstreamProvider):
|
||||
@override
|
||||
def match_source(
|
||||
self,
|
||||
source: str,
|
||||
current_version: str,
|
||||
) -> GiteaTagSource | None:
|
||||
parsed = urllib.parse.urlparse(source)
|
||||
if parsed.scheme not in {"http", "https"} or parsed.netloc == "github.com":
|
||||
return None
|
||||
|
||||
parts = urllib.parse.unquote(parsed.path).strip("/").split("/")
|
||||
if len(parts) < GITEA_ARCHIVE_PATH_PART_COUNT or parts[2] != "archive":
|
||||
return None
|
||||
|
||||
tag = archive_tag_name(parts[3])
|
||||
if tag is None:
|
||||
return None
|
||||
|
||||
tag_template = release_tag_template(tag, current_version)
|
||||
if tag_template is None:
|
||||
return None
|
||||
|
||||
return GiteaTagSource(
|
||||
scheme=parsed.scheme,
|
||||
host=parsed.netloc,
|
||||
owner=parts[0],
|
||||
repo=parts[1],
|
||||
tag_template=tag_template,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _api_headers() -> dict[str, str]:
|
||||
headers = {
|
||||
"Accept": "application/json",
|
||||
"User-Agent": "pkg-sync",
|
||||
}
|
||||
token = os.environ.get("GITEA_TOKEN")
|
||||
if token:
|
||||
headers["Authorization"] = f"token {token}"
|
||||
return headers
|
||||
|
||||
@staticmethod
|
||||
def _tags_url(source: GiteaTagSource) -> str:
|
||||
owner = urllib.parse.quote(source.owner, safe="")
|
||||
repo = urllib.parse.quote(source.repo, safe="")
|
||||
return (
|
||||
f"{source.scheme}://{source.host}/api/v1/repos/"
|
||||
f"{owner}/{repo}/tags?limit=50"
|
||||
)
|
||||
|
||||
def _tag_names(self, source: GiteaTagSource) -> list[str] | None:
|
||||
request = urllib.request.Request(
|
||||
self._tags_url(source),
|
||||
headers=self._api_headers(),
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=60) as response:
|
||||
data = json.load(response)
|
||||
except (OSError, urllib.error.URLError, ValueError) as exc:
|
||||
error(
|
||||
f"failed to check Gitea tags for "
|
||||
f"{source.host}/{source.owner}/{source.repo}: {exc}"
|
||||
)
|
||||
return None
|
||||
|
||||
if not isinstance(data, list):
|
||||
error(
|
||||
f"Gitea tags for {source.host}/{source.owner}/{source.repo} "
|
||||
"did not return a list"
|
||||
)
|
||||
return None
|
||||
|
||||
names: list[str] = []
|
||||
for item in data:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
name = item.get("name")
|
||||
if isinstance(name, str) and name:
|
||||
names.append(name)
|
||||
|
||||
if not names:
|
||||
error(
|
||||
f"Gitea tags for "
|
||||
f"{source.host}/{source.owner}/{source.repo} are empty"
|
||||
)
|
||||
return None
|
||||
return names
|
||||
|
||||
@override
|
||||
def latest_version(self, source: ReleaseSource) -> str | None:
|
||||
if not isinstance(source, GiteaTagSource):
|
||||
raise TypeError("Gitea provider received non-Gitea source")
|
||||
|
||||
tags = self._tag_names(source)
|
||||
if tags is None:
|
||||
return None
|
||||
|
||||
versions: list[str] = []
|
||||
for tag in tags:
|
||||
version = version_from_tag(source, tag, report_mismatch=False)
|
||||
if version is not None:
|
||||
versions.append(version)
|
||||
if not versions:
|
||||
error(
|
||||
f"Gitea tags for {source.host}/{source.owner}/{source.repo} "
|
||||
f"have no tags matching {source.tag_template!r}"
|
||||
)
|
||||
return None
|
||||
|
||||
return max(versions, key=version_sort_key)
|
||||
|
||||
|
||||
UPSTREAM_PROVIDERS: Final[tuple[UpstreamProvider, ...]] = (
|
||||
GithubProvider(),
|
||||
GiteaProvider(),
|
||||
)
|
||||
|
||||
|
||||
def upstream_match_from_source(
|
||||
source: str,
|
||||
current_version: str,
|
||||
) -> UpstreamMatch | None:
|
||||
for provider in UPSTREAM_PROVIDERS:
|
||||
release_source = provider.match_source(source, current_version)
|
||||
if release_source is not None:
|
||||
return UpstreamMatch(provider=provider, source=release_source)
|
||||
return None
|
||||
|
||||
|
||||
def upstream_match_from_sources(
|
||||
sources: tuple[str, ...],
|
||||
current_version: str,
|
||||
) -> UpstreamMatch | None:
|
||||
for source in sources:
|
||||
upstream_match = upstream_match_from_source(source, current_version)
|
||||
if upstream_match is not None:
|
||||
return upstream_match
|
||||
return None
|
||||
|
||||
|
||||
def rpm_upstream_match(spec: Path, current_version: str) -> UpstreamMatch | None:
|
||||
parsed_sources = rpm_spec_sources(spec)
|
||||
if parsed_sources is None:
|
||||
return None
|
||||
|
||||
return upstream_match_from_sources(parsed_sources, current_version)
|
||||
|
||||
|
||||
def rpm_file_metadata(path: Path) -> RpmPackage | None:
|
||||
output = command_output(
|
||||
[
|
||||
@@ -969,7 +1177,7 @@ class OpenSuseBackend(Backend[RpmTarget]):
|
||||
def required_commands(self) -> Sequence[str]:
|
||||
commands = ["rpmbuild", "rpmspec", "rpm"]
|
||||
if self.options.source_fetch:
|
||||
commands.append("rpmdev-spectool")
|
||||
commands.extend(["rpmdev-bumpspec", "rpmdev-spectool"])
|
||||
return tuple(commands)
|
||||
|
||||
@override
|
||||
@@ -1028,18 +1236,6 @@ class OpenSuseBackend(Backend[RpmTarget]):
|
||||
groups.setdefault(target.package_dir, []).append(target)
|
||||
return list(groups.values())
|
||||
|
||||
@staticmethod
|
||||
def _github_release_source(spec: Path) -> GithubReleaseSource | None:
|
||||
sources = rpm_raw_spec_sources(spec)
|
||||
if sources is None:
|
||||
return None
|
||||
|
||||
for source in sources:
|
||||
release_source = github_release_source(source)
|
||||
if release_source is not None:
|
||||
return release_source
|
||||
return None
|
||||
|
||||
def _source_dir(self, target: RpmTarget) -> Path:
|
||||
return self._rpmbuild_topdir(target) / "SOURCES"
|
||||
|
||||
@@ -1117,32 +1313,26 @@ class OpenSuseBackend(Backend[RpmTarget]):
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _latest_github_version(
|
||||
release_source: GithubReleaseSource,
|
||||
def _latest_upstream_version(
|
||||
upstream_match: UpstreamMatch,
|
||||
) -> str | None:
|
||||
tag = github_latest_release_tag(release_source)
|
||||
if tag is None:
|
||||
return None
|
||||
return github_version_from_tag(release_source, tag)
|
||||
return upstream_match.provider.latest_version(upstream_match.source)
|
||||
|
||||
@staticmethod
|
||||
def _updated_spec_sources(
|
||||
target: RpmTarget, version: str
|
||||
) -> tuple[str, ...] | None:
|
||||
if not rpm_rewrite_spec_version(target.spec, version):
|
||||
if not rpm_bump_spec_version(target.spec, version):
|
||||
return None
|
||||
return rpm_spec_sources(target.spec)
|
||||
|
||||
def _apply_github_update(
|
||||
def _apply_release_update(
|
||||
self,
|
||||
target: RpmTarget,
|
||||
release_source: GithubReleaseSource,
|
||||
upstream_match: UpstreamMatch,
|
||||
current: str,
|
||||
) -> UpstreamUpdateResult:
|
||||
current = rpm_spec_version(target.spec)
|
||||
if current is None:
|
||||
return UpstreamUpdateResult.FAILED
|
||||
|
||||
latest = self._latest_github_version(release_source)
|
||||
latest = self._latest_upstream_version(upstream_match)
|
||||
if latest is None:
|
||||
return UpstreamUpdateResult.FAILED
|
||||
|
||||
@@ -1160,11 +1350,15 @@ class OpenSuseBackend(Backend[RpmTarget]):
|
||||
if not self.options.source_fetch:
|
||||
return UpstreamUpdateResult.UNCHANGED
|
||||
|
||||
release_source = self._github_release_source(target.spec)
|
||||
if release_source is None:
|
||||
current = rpm_spec_version(target.spec)
|
||||
if current is None:
|
||||
return UpstreamUpdateResult.FAILED
|
||||
|
||||
upstream_match = rpm_upstream_match(target.spec, current)
|
||||
if upstream_match is None:
|
||||
return UpstreamUpdateResult.UNCHANGED
|
||||
|
||||
return self._apply_github_update(target, release_source)
|
||||
return self._apply_release_update(target, upstream_match, current)
|
||||
|
||||
def _prepare_rpmbuild_tree(self, target: RpmTarget) -> bool:
|
||||
topdir = self._rpmbuild_topdir(target)
|
||||
|
||||
Reference in New Issue
Block a user