mrtest: Add support for using an MR as a repo when upgrading

This is useful to test how apk actually will behave when upgrading.

mrhlpr/mr.py

@@ -44,6 +44,12 @@ class MergeRequestStatus:
     state: str
 
 
+@dataclass
+class PipelineMetadata:
+    host_arch_job: dict
+    project_id: int
+
+
 def get_status(
     mr_id: int, no_cache: bool = False, origin: Optional[gitlab.GitLabOrigin] = None
 ) -> MergeRequestStatus:
@@ -119,14 +125,7 @@ def get_status(
     )
 
 
-def get_artifacts_zip(mr_id, no_cache=False, origin):
-    """Download artifacts from the GitLab API.
-
-    :param mr_id: merge request ID
-    :param no_cache: do not cache the API result for the merge request data
-    :param origin: gitlab origin information, see gitlab.parse_git_origin()
-    :returns: path to downloaded artifacts.zip file
-    """
+def get_pipeline_metadata(mr_id, no_cache: bool, origin: gitlab.GitLabOrigin) -> PipelineMetadata:
     # Get the latest pipeline (without cache so we don't miss newer ones)
     url_mr = "/projects/{}/merge_requests/{}".format(origin.api_project_id, mr_id)
     api = gitlab.download_json(origin, url_mr, no_cache=True)
@@ -155,8 +154,67 @@ def get_artifacts_zip(mr_id, no_cache=False, origin):
         )
         exit(1)
 
+    return PipelineMetadata(host_arch_job=job, project_id=pipeline_project_id)
+
+
+def get_artifacts_zip(mr_id: int, no_cache: bool, origin: gitlab.GitLabOrigin) -> str:
+    """Download artifacts from the GitLab API.
+
+    :param mr_id: merge request ID
+    :param no_cache: do not cache the API result for the merge request data
+    :param origin: gitlab origin information, see gitlab.parse_git_origin()
+    :returns: path to downloaded artifacts.zip file
+    """
+    pipeline_metadata = get_pipeline_metadata(mr_id, no_cache, origin)
     # Download artifacts zip (with cache)
-    return gitlab.download_artifacts_zip(origin.api, pipeline_project_id, job)
+    return gitlab.download_artifacts_zip(
+        origin.api,
+        pipeline_metadata.project_id,
+        pipeline_metadata.host_arch_job,
+    )
+
+
+class UnknownReleaseError(ValueError): ...
+
+
+def _target_branch_to_pmos_release(target_branch: str) -> str:
+    if target_branch == "master":
+        return "edge"
+
+    pattern = re.compile(r"^v\d\d\.\d\d$")
+
+    if pattern.match(target_branch):
+        return target_branch
+
+    raise UnknownReleaseError
+
+
+def get_artifacts_repo_urls(
+    mr_id: int, no_cache: bool, origin: gitlab.GitLabOrigin, alpine_mr: bool
+) -> list[str]:
+    """Get a list of URLs that can be used as repositories for apk. These URLs contain
+    the apks built as part of the latest job for the host system's architecture in the
+    given MR."""
+    pipeline_metadata = get_pipeline_metadata(mr_id, no_cache, origin)
+    url_mr = "/projects/{}/merge_requests/{}".format(
+        origin.api_project_id,
+        mr_id,
+    )
+    api = gitlab.download_json(origin, url_mr, no_cache=True)
+
+    host_arch_job_web_url = pipeline_metadata.host_arch_job["web_url"]
+    if alpine_mr:
+        return [
+            f"{host_arch_job_web_url}/artifacts/raw/packages/main",
+            f"{host_arch_job_web_url}/artifacts/raw/packages/community",
+            f"{host_arch_job_web_url}/artifacts/raw/packages/testing",
+        ]
+    else:
+        pmos_release = _target_branch_to_pmos_release(api["target_branch"])
+
+        return [
+            f"{host_arch_job_web_url}/artifacts/raw/packages/{pmos_release}",
+        ]
 
 
 def checkout(

mrtest/add_packages.py

@@ -7,6 +7,7 @@ import os
 import shutil
 import zipfile
 import subprocess
+from typing import Literal
 from urllib.error import HTTPError
 
 import mrhlpr.mr
@@ -68,7 +69,7 @@ def run_apk_add(origin, mr_id, apk_paths):
     subprocess.run(cmd, check=True)
 
 
-def confirm_mr_id(origin, mr_id):
+def confirm_mr_id(origin, mr_id: int, action: Literal["add", "upgrade"]) -> None:
     """
     :param origin: gitlab origin information, see gitlab.parse_git_origin()
     :param mr_id: merge request ID
@@ -76,6 +77,7 @@ def confirm_mr_id(origin, mr_id):
     link = f"https://{origin.host}/{origin.project_id}/-/merge_requests/{mr_id}"
 
     status = mrhlpr.mr.get_status(mr_id, origin=origin)
+    action_description = "select and then install" if action == "add" else "upgrade"
 
     print("Welcome to mrtest, this tool allows downloading and installing")
     print("Alpine packages from merge requests.")
@@ -85,7 +87,7 @@ def confirm_mr_id(origin, mr_id):
     print("Malicious code may make your device permanently unusable, steal")
     print("your passwords, and worse.")
     print()
-    print("You are about to select and then install packages from:")
+    print(f"You are about to {action_description} packages from:")
     print(link)
     print("\t“\033[1m" + status.title + "\033[m”")
     print("\t(\033[3m" + status.source + ":" + status.source_branch + "\033[m)")
@@ -111,7 +113,7 @@ def add_packages(origin, mr_id, no_cache):
     :param no_cache: instead of using a cache for api calls / downloads where
                      it makes sense, always download a fresh copy
     """
-    confirm_mr_id(origin, mr_id)
+    confirm_mr_id(origin, mr_id, "add")
 
     try:
         zip_path = mrhlpr.mr.get_artifacts_zip(mr_id, no_cache, origin)

mrtest/frontend.py

@@ -8,6 +8,7 @@ import sys
 
 import mrtest.add_packages
 import mrtest.origin
+import mrtest.upgrade_packages
 import mrtest.zap_packages
 
 try:
@@ -18,13 +19,28 @@ except ImportError:
 
 def parse_args_parser_add(sub):
     """:param sub: argparser's subparser"""
-    parser = sub.add_parser("add", help="install/upgrade to packages from a MR")
+    parser = sub.add_parser("add", help="install packages from an MR")
     parser.add_argument(
         "-a", "--alpine", action="store_true", help="use alpine's aports instead of pmOS' pmaports"
     )
     parser.add_argument("mr_id", type=int, help="merge request ID")
 
 
+def parse_args_parser_upgrade(sub) -> None:
+    """:param sub: argparser's subparser"""
+    parser = sub.add_parser("upgrade", help="upgrade to packages from an MR")
+    parser.add_argument(
+        "-a", "--alpine", action="store_true", help="use alpine's aports instead of pmOS' pmaports"
+    )
+    parser.add_argument(
+        "-l",
+        "--available",
+        action="store_true",
+        help="install remote versions of packages even if they aren't a higher version",
+    )
+    parser.add_argument("mr_id", type=int, help="merge request ID")
+
+
 def parse_args_parser_zap(sub):
     """:param sub: argparser's subparser"""
     sub.add_parser("zap", help="uninstall previously added packages")
@@ -45,6 +61,7 @@ def parse_args():
     sub.required = True
 
     parse_args_parser_add(sub)
+    parse_args_parser_upgrade(sub)
     parse_args_parser_zap(sub)
 
     if "argcomplete" in sys.modules:
@@ -56,8 +73,10 @@ def main():
     args = parse_args()
     if args.verbose:
         logging.getLogger().setLevel(logging.DEBUG)
+    origin = mrtest.origin.aports if args.alpine else mrtest.origin.pmaports
     if args.action == "add":
-        origin = mrtest.origin.aports if args.alpine else mrtest.origin.pmaports
         mrtest.add_packages.add_packages(origin, args.mr_id, args.no_cache)
+    elif args.action == "upgrade":
+        mrtest.upgrade_packages.upgrade_from_mr(origin, args.mr_id, args.alpine, args.available)
     elif args.action == "zap":
         mrtest.zap_packages.zap_packages()

mrtest/upgrade_packages.py

+# Copyright 2024 Stefan Hansson
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+import logging
+import subprocess
+
+import mrtest
+from mrhlpr.gitlab import GitLabOrigin
+from mrhlpr.mr import get_artifacts_repo_urls
+from mrtest.add_packages import confirm_mr_id
+
+
+def upgrade_from_mr(origin: GitLabOrigin, mr_id: int, alpine_mr: bool, available: bool) -> None:
+    confirm_mr_id(origin, mr_id, "upgrade")
+    repo_urls = get_artifacts_repo_urls(mr_id, True, origin, alpine_mr)
+
+    repo_args = []
+
+    for repo_url in repo_urls:
+        repo_args += ["-X", repo_url]
+
+    cmd = [
+        "apk",
+        "upgrade",
+        *repo_args,
+        "--allow-untrusted",
+        "--force-missing-repositories",
+        "--interactive",
+    ]
+
+    if available:
+        cmd.append("--available")
+
+    if not mrtest.is_root_user():  # type: ignore[attr-defined]
+        cmd = [mrtest.get_sudo()] + cmd  # type: ignore[attr-defined]
+
+    print("Upgrading packages...")
+    logging.debug(f"+ {cmd}")
+    subprocess.run(cmd, check=True)