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

Should be okay now.

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(