feat(RELEASE-2533): add update_infra_deployments.py script (konflux-ci#803)

This commit adds a python script to replicate the functionality of the
inline bash script in the update-infra-deployments task in the catalog
repo. The task will be updated to use this python module instead.

Assisted-By: Cursor

Signed-off-by: Johnny Bieren <jbieren@redhat.com>

Author: johnbieren

scripts/python/helpers/file.py

@@ -3,9 +3,21 @@
 from __future__ import annotations
 
 import hashlib
+import json
 import os
 import tempfile
 from pathlib import Path
+from typing import Any
+
+
+def load_json_dict(path: Path) -> dict[str, Any]:
+    """Load a JSON file whose root value must be an object."""
+    with open(path, encoding="utf-8") as handle:
+        data = json.load(handle)
+    if not isinstance(data, dict):
+        msg = f"JSON root must be an object: {path}"
+        raise TypeError(msg)
+    return data
 
 
 def sha256(path: Path) -> str:
@@ -21,14 +33,14 @@ def path_from_env_variable(
     name: str,
     default: str | Path,
 ) -> Path:
-    """Return a filesystem path from the named environment variable, or *default*.
+    """Return a filesystem path from an environment variable, or a default.
 
-    The value of *name* in ``os.environ`` (if set and non-blank after
-    ``str.strip``) is interpreted as a path; this function does not open or
-    stat paths.
+    The value of name in `os.environ` (if set and not blank after
+    `str.strip`) is interpreted as a path; it is not a path to a file whose
+    contents you read, and this function does not open or stat paths.
 
-    If the variable is missing or only whitespace, *default* is returned (a
-    str or an existing ``Path``).
+    If the variable is missing or only whitespace, default is returned (a str
+    or an existing `Path`).
 
     Typical use: a Tekton or pod env var that holds a mount directory path, with
     tests setting the same variable to a temp directory. Existence of the path
@@ -46,13 +58,13 @@ def make_tempfile_path(
 ) -> Path:
     """Create a secure private temp file and return a pathlib.Path to it.
 
-    Uses the standard library ``tempfile.mkstemp``, which creates a new file and
-    returns a file handle (a safe pattern). We never use the old ``mktemp`` API
-    (unsafe under concurrency; deprecated in Python 3.12). If ``data`` is given,
+    Uses the standard library `tempfile.mkstemp`, which creates a new file and
+    returns a file handle (a safe pattern). We never use the old `mktemp` API
+    (unsafe under concurrency; deprecated in Python 3.12). If `data` is given,
     those bytes are written into the new file; otherwise the file is empty. The
     file is closed before returning; the caller is responsible for deleting the
-    path when done. ``prefix`` is the filename prefix in the system temp
-    directory, same as the ``prefix`` argument to ``mkstemp``.
+    path when done. `prefix` is the filename prefix in the system temp
+    directory, same as the `prefix` argument to `mkstemp`.
     """
     fd, name = tempfile.mkstemp(prefix=prefix)
     try:

scripts/python/helpers/image_ref.py

@@ -2,10 +2,56 @@
 
 from __future__ import annotations
 
+import json
+import re
 
-def pyxis_url_for_pull_spec(pyxis_url: str, pull_spec: str) -> str:
+import http_client
+import requests
+
+_QUAY_SHA_TAG = re.compile(r"^[0-9a-f]{40}$")
+_MAX_QUAY_TAG_PAGES = 50
+
+
+def resolve_quay_digest_to_git_sha(digest: str, container_image: str) -> str | None:
+    """Resolve an image digest to a git commit SHA via the Quay public API.
+
+    Returns `None` when resolution fails (non-quay image, no matching tag, etc).
     """
-    Build the Pyxis repository/tag API URL for `pull_spec`.
+    try:
+        repo_url = container_image.split("@", 1)[0]
+        if not repo_url.startswith("quay.io/"):
+            print("Not a quay.io image, skipping digest resolution")
+            return None
+        repo_path = repo_url.removeprefix("quay.io/")
+        page = 1
+        while page <= _MAX_QUAY_TAG_PAGES:
+            url = (
+                f"https://quay.io/api/v1/repository/{repo_path}/tag/" f"?limit=100&page={page}"
+            )
+            try:
+                body = http_client.get_text(url, timeout=10)
+            except requests.HTTPError as exc:
+                code = exc.response.status_code if exc.response is not None else "?"
+                print(f"Quay API returned {code}, skipping digest resolution")
+                return None
+            data = json.loads(body)
+            for tag in data.get("tags", []):
+                name = tag.get("name", "")
+                if tag.get("manifest_digest") == digest and _QUAY_SHA_TAG.fullmatch(name):
+                    print(f"Resolved {digest[:19]}... to git SHA {name}")
+                    return str(name)
+            if not data.get("has_additional", False):
+                break
+            page += 1
+        print(f"No git SHA tag found for digest {digest[:19]}...")
+        return None
+    except Exception as exc:
+        print(f"Failed to resolve digest to git SHA: {exc}")
+        return None
+
+
+def pyxis_url_for_pull_spec(pyxis_url: str, pull_spec: str) -> str:
+    """Build the Pyxis repository/tag API URL for `pull_spec`.
 
     `registry.redhat.io` is rewritten to `registry.access.redhat.com` to
     match Pyxis lookups.

scripts/python/helpers/snapshot.py

@@ -0,0 +1,40 @@
+"""Helpers for reading Konflux snapshot JSON documents."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import file
+
+
+def first_component(snapshot_path: Path) -> dict[str, str]:
+    """Return fields from the first snapshot `components` entry.
+
+    Keys: `revision`, `origin_repo` (`source.git.url` without `.git`),
+    and `container_image`.
+    """
+    snapshot = file.load_json_dict(snapshot_path)
+    components = snapshot.get("components")
+    if not isinstance(components, list) or not components:
+        msg = f"snapshot has no components: {snapshot_path}"
+        raise ValueError(msg)
+    first = components[0]
+    if not isinstance(first, dict):
+        msg = f"snapshot component[0] must be an object: {snapshot_path}"
+        raise TypeError(msg)
+    source = first.get("source")
+    if not isinstance(source, dict):
+        msg = f"snapshot component[0].source must be an object: {snapshot_path}"
+        raise TypeError(msg)
+    git_info = source.get("git")
+    if not isinstance(git_info, dict):
+        msg = f"snapshot component[0].source.git must be an object: {snapshot_path}"
+        raise TypeError(msg)
+    revision = str(git_info.get("revision", "")).strip()
+    origin_repo = str(git_info.get("url", "")).strip().removesuffix(".git")
+    container_image = str(first.get("containerImage", "")).strip()
+    return {
+        "revision": revision,
+        "origin_repo": origin_repo,
+        "container_image": container_image,
+    }

scripts/python/helpers/test_file.py

@@ -1,7 +1,8 @@
-"""Tests for the ``file`` helper module."""
+"""Tests for the `file` helper module."""
 
 from __future__ import annotations
 
+import json
 from pathlib import Path
 
 import file
@@ -11,7 +12,7 @@
 def test_path_from_env_variable_uses_set_value(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
-    """A non-empty env value (after trim) is returned as a ``Path``."""
+    """A non-empty env value (after trim) is returned as a `Path`."""
     p = tmp_path / "m"
     monkeypatch.setenv("MOUNT", str(p))
     assert file.path_from_env_variable("MOUNT", "/d/e/f") == p
@@ -29,7 +30,7 @@ def test_path_from_env_variable_strips_value(
 def test_path_from_env_variable_uses_default_when_unset(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
-    """Unset or all-whitespace *name* yields *default* (``Path`` or str)."""
+    """Unset or all-whitespace *name* yields *default* (`Path` or str)."""
     default = str(tmp_path / "default")
     monkeypatch.delenv("MOUNT", raising=False)
     assert file.path_from_env_variable("MOUNT", default) == tmp_path / "default"
@@ -40,14 +41,29 @@ def test_path_from_env_variable_uses_default_when_unset(
 def test_path_from_env_variable_path_default(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
-    """*default* may be a ``Path`` object, returned unchanged when the env is unset."""
+    """*default* may be a `Path` object, returned unchanged when the env is unset."""
     d = tmp_path / "d"
     monkeypatch.delenv("MOUNTX", raising=False)
     assert file.path_from_env_variable("MOUNTX", d) == d
 
 
+def test_load_json_dict(tmp_path: Path) -> None:
+    """A JSON object file is parsed and returned as a dict."""
+    path = tmp_path / "data.json"
+    path.write_text(json.dumps({"a": 1}), encoding="utf-8")
+    assert file.load_json_dict(path) == {"a": 1}
+
+
+def test_load_json_dict_rejects_non_object(tmp_path: Path) -> None:
+    """A JSON array (non-object root) raises `TypeError`."""
+    path = tmp_path / "data.json"
+    path.write_text("[1]", encoding="utf-8")
+    with pytest.raises(TypeError, match="object"):
+        file.load_json_dict(path)
+
+
 def test_make_tempfile_path_empty_file() -> None:
-    """A ``None`` payload leaves the created file with zero length."""
+    """A `None` payload leaves the created file with zero length."""
     p = file.make_tempfile_path("t-", None)
     try:
         assert p.read_bytes() == b""
@@ -56,7 +72,7 @@ def test_make_tempfile_path_empty_file() -> None:
 
 
 def test_make_tempfile_path_with_bytes() -> None:
-    """If ``data`` is set, the file on disk has exactly those bytes."""
+    """If `data` is set, the file on disk has exactly those bytes."""
     p = file.make_tempfile_path("t-", b"hello")
     try:
         assert p.read_bytes() == b"hello"

scripts/python/helpers/test_image_ref.py

@@ -2,8 +2,12 @@
 
 from __future__ import annotations
 
+import json
+from unittest import mock
+
 import image_ref
 import pytest
+import requests
 
 
 def test_pyxis_url_for_pull_spec_with_tag_and_registry_rewrite() -> None:
@@ -28,3 +32,103 @@ def test_pyxis_url_for_pull_spec_invalid() -> None:
     """Invalid pull specs raise `ValueError`."""
     with pytest.raises(ValueError, match="invalid pull spec"):
         image_ref.pyxis_url_for_pull_spec("https://pyxis/v1", "not/a-pullspec")
+
+
+def test_resolve_quay_digest_skips_non_quay() -> None:
+    """Non-quay.io images return `None` without calling the Quay API."""
+    assert (
+        image_ref.resolve_quay_digest_to_git_sha(
+            "sha256:abc",
+            "registry.io/org/repo@sha256:abc",
+        )
+        is None
+    )
+
+
+def test_resolve_quay_digest_finds_sha_tag() -> None:
+    """A 40-char hex tag matching the digest is returned from the first API page."""
+    digest = "sha256:" + "a" * 64
+    sha = "b" * 40
+    payload = json.dumps(
+        {
+            "tags": [{"name": sha, "manifest_digest": digest}],
+            "has_additional": False,
+        }
+    )
+    with mock.patch("image_ref.http_client.get_text", return_value=payload):
+        out = image_ref.resolve_quay_digest_to_git_sha(
+            digest,
+            f"quay.io/org/repo@{digest}",
+        )
+    assert out == sha
+
+
+def test_resolve_quay_digest_non_200_response() -> None:
+    """Quay API errors return `None` instead of raising."""
+    digest = "sha256:" + "a" * 64
+    response = mock.MagicMock(status_code=503)
+    with mock.patch(
+        "image_ref.http_client.get_text",
+        side_effect=requests.HTTPError(response=response),
+    ):
+        out = image_ref.resolve_quay_digest_to_git_sha(
+            digest,
+            f"quay.io/org/repo@{digest}",
+        )
+    assert out is None
+
+
+def test_resolve_quay_digest_paginates() -> None:
+    """Resolution follows `has_additional` across multiple tag-list pages."""
+    digest = "sha256:" + "a" * 64
+    sha = "c" * 40
+    page_one = json.dumps({"tags": [], "has_additional": True})
+    page_two = json.dumps(
+        {
+            "tags": [{"name": sha, "manifest_digest": digest}],
+            "has_additional": False,
+        }
+    )
+    with mock.patch(
+        "image_ref.http_client.get_text",
+        side_effect=[page_one, page_two],
+    ) as get_text:
+        out = image_ref.resolve_quay_digest_to_git_sha(
+            digest,
+            f"quay.io/org/repo@{digest}",
+        )
+    assert out == sha
+    assert get_text.call_count == 2
+
+
+def test_resolve_quay_digest_no_matching_tag() -> None:
+    """Return `None` when no tag has both the digest and a 40-char hex name."""
+    digest = "sha256:" + "a" * 64
+    payload = json.dumps(
+        {
+            "tags": [
+                {"name": "not-a-sha", "manifest_digest": digest},
+                {"name": "b" * 40, "manifest_digest": "sha256:other"},
+            ],
+            "has_additional": False,
+        }
+    )
+    with mock.patch("image_ref.http_client.get_text", return_value=payload):
+        out = image_ref.resolve_quay_digest_to_git_sha(
+            digest,
+            f"quay.io/org/repo@{digest}",
+        )
+    assert out is None
+
+
+def test_resolve_quay_digest_handles_exception() -> None:
+    """Unexpected failures are swallowed and return `None`."""
+    with mock.patch(
+        "image_ref.http_client.get_text",
+        side_effect=RuntimeError("network down"),
+    ):
+        out = image_ref.resolve_quay_digest_to_git_sha(
+            "sha256:abc",
+            "quay.io/org/repo@sha256:abc",
+        )
+    assert out is None