Fix Release bumping for RHEL in maintenance phase

ymir/common/tests/unit/test_version_utils.py

@@ -1,6 +1,8 @@
 import pytest
+from flexmock import flexmock
 
 from ymir.common.version_utils import (
+    get_maintenance_rhel_branch,
     is_older_zstream,
     parse_branch_name,
     parse_rhel_version,
@@ -135,3 +137,35 @@ async def test_is_older_zstream(version_or_branch, expected):
     }
     result = await is_older_zstream(version_or_branch, CURRENT_Z_STREAMS)
     assert result == expected
+
+
+RHEL_CONFIG = {
+    "current_y_streams": {"9": "rhel-9.8", "10": "rhel-10.2"},
+    "current_z_streams": {"8": "rhel-8.10.z", "9": "rhel-9.7.z", "10": "rhel-10.1.z"},
+}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "branch, expected",
+    [
+        ("c8s", "rhel-8.10.0"),
+        ("c9s", None),
+        ("c10s", None),
+        ("rhel-8.10.0", None),
+        ("rhel-8-main", None),
+        ("rhel-9.7.0", None),
+        ("rhel-10-main", None),
+        ("invalid", None),
+        ("", None),
+    ],
+)
+async def test_get_maintenance_rhel_branch(branch, expected):
+    from ymir.common import config
+
+    async def mock_load_rhel_config():
+        return RHEL_CONFIG
+
+    flexmock(config).should_receive("load_rhel_config").replace_with(mock_load_rhel_config)
+    result = await get_maintenance_rhel_branch(branch)
+    assert result == expected

ymir/common/version_utils.py

@@ -125,6 +125,27 @@ def get_maintenance_majors(rhel_config: dict) -> set[str]:
     return set(current_z_streams.keys()) - set(current_y_streams.keys())
 
 
+async def get_maintenance_rhel_branch(branch: str) -> str | None:
+    """Get internal maintenance phase RHEL branch corresponding to the given CentOS Stream branch, if any."""
+    from ymir.common.base_utils import is_cs_branch
+    from ymir.common.config import load_rhel_config
+
+    if not is_cs_branch(branch):
+        return None
+    if not (parsed := parse_branch_name(branch)):
+        return None
+    major, _ = parsed
+
+    config = await load_rhel_config()
+    if major not in get_maintenance_majors(config):
+        return None
+    z_stream = config.get("current_z_streams", {}).get(major)
+    if not z_stream or not (z_parsed := parse_rhel_version(z_stream)):
+        return None
+    z_major, z_minor, _ = z_parsed
+    return construct_internal_branch_name(z_major, z_minor)
+
+
 def get_fix_version_variants(fix_version: str) -> list[str]:
     """
     Return both Y-stream and Z-stream forms for a given fixVersion.

ymir/tools/unprivileged/specfile.py

@@ -23,6 +23,7 @@
 )
 
 from ymir.common.utils import get_absolute_path, get_all_patches, get_latest_candidate_build
+from ymir.common.version_utils import get_maintenance_rhel_branch
 from ymir.tools.base import CloneableTool as Tool
 
 logger = logging.getLogger(__name__)
@@ -192,9 +193,13 @@ class UpdateReleaseTool(Tool[UpdateReleaseToolInput, ToolRunOptions, StringToolO
     description = """
     Updates the value of the `Release` field in the specified spec file.
 
-    If branch is a Z-Stream branch (rhel-X.Y or rhel-X.Y.Z), release is updated in the following way:
-        - base release is established - from the latest current stream candidate build unless the latest
-          higher stream (Y + 1) candidate build shares the same version but has a higher release
+    If branch is a Z-Stream branch (rhel-X.Y or rhel-X.Y.Z) or a CentOS Stream branch for a
+    RHEL version in maintenance phase (e.g. c8s), release is updated in the following way:
+        - base release is established - from the latest candidate build of the current stream (for
+          CentOS Stream branches corresponding to a RHEL version in maintenance phase, the internal
+          RHEL branch is used for the candidate build lookup), unless the latest higher stream (Y + 1)
+          candidate build shares the same version but has a higher release (not applicable to
+          maintenance phase RHEL as there is no higher stream)
         - if %autorelease is present in the current Release:
             - if abandon_autorelease is True, %autorelease is removed and Release is set to
               "N%{?dist}.1" (or "0%{?dist}.1" for rebase), using a plain numeric Z-stream counter
@@ -293,7 +298,7 @@ async def _set_zstream_release(
         package: str,
         rebase: bool,
         current_stream_branch: str,
-        higher_stream_branch: str,
+        higher_stream_branch: str | None = None,
         abandon_autorelease: bool = False,
     ) -> None:
         def extract_numeric_release(evr):
@@ -313,18 +318,25 @@ def extract_zstream_suffix(evr):
                     return 0
             return 0
 
-        (latest_current_stream_build, _), (latest_higher_stream_build, _) = await asyncio.gather(
-            get_latest_candidate_build(package, current_stream_branch),
-            get_latest_candidate_build(package, higher_stream_branch),
-        )
-        base_build = (
-            latest_current_stream_build
-            if EVR(epoch=latest_current_stream_build.epoch, version=latest_current_stream_build.version)
-            != EVR(epoch=latest_higher_stream_build.epoch, version=latest_higher_stream_build.version)
-            or extract_numeric_release(latest_current_stream_build)
-            > extract_numeric_release(latest_higher_stream_build)
-            else latest_higher_stream_build
-        )
+        if higher_stream_branch:
+            (latest_current_stream_build, _), (latest_higher_stream_build, _) = await asyncio.gather(
+                get_latest_candidate_build(package, current_stream_branch),
+                get_latest_candidate_build(package, higher_stream_branch),
+            )
+            base_build = latest_current_stream_build
+            if EVR(
+                epoch=latest_higher_stream_build.epoch,
+                version=latest_higher_stream_build.version,
+            ) == EVR(
+                epoch=latest_current_stream_build.epoch,
+                version=latest_current_stream_build.version,
+            ) and extract_numeric_release(latest_higher_stream_build) >= extract_numeric_release(
+                latest_current_stream_build
+            ):
+                base_build = latest_higher_stream_build
+        else:
+            latest_current_stream_build, _ = await get_latest_candidate_build(package, current_stream_branch)
+            base_build = latest_current_stream_build
         base_release = extract_numeric_release(base_build)
         with Specfile(spec_path) as spec:
             current_release = spec.raw_release
@@ -386,17 +398,21 @@ async def _run(
     ) -> StringToolOutput:
         spec_path = get_absolute_path(tool_input.spec, self)
         try:
-            if not (higher_stream_branch := self._get_higher_stream_branch(tool_input.dist_git_branch)):
-                await self._bump_or_reset_release(spec_path, tool_input.rebase)
-            else:
+            higher_stream_branch = self._get_higher_stream_branch(tool_input.dist_git_branch)
+            maintenance_rhel_branch = not higher_stream_branch and await get_maintenance_rhel_branch(
+                tool_input.dist_git_branch
+            )
+            if higher_stream_branch or maintenance_rhel_branch:
                 await self._set_zstream_release(
                     spec_path,
                     tool_input.package,
                     tool_input.rebase,
-                    tool_input.dist_git_branch,
+                    maintenance_rhel_branch or tool_input.dist_git_branch,
                     higher_stream_branch,
                     abandon_autorelease=tool_input.abandon_autorelease,
                 )
+            else:
+                await self._bump_or_reset_release(spec_path, tool_input.rebase)
         except Exception as e:
             raise ToolError(f"Failed to update release: {e}") from e
         return StringToolOutput(result=f"Successfully updated release in {spec_path}")

ymir/tools/unprivileged/tests/unit/test_specfile.py

@@ -425,6 +425,13 @@ async def mock_get_latest_candidate_build(package, dist_git_branch_arg):
         mock_get_latest_candidate_build
     )
 
+    async def mock_get_maintenance_rhel_branch(_):
+        return None
+
+    flexmock(specfile_tools).should_receive("get_maintenance_rhel_branch").replace_with(
+        mock_get_maintenance_rhel_branch
+    )
+
     tool = UpdateReleaseTool()
 
     async def run_and_check(spec, expected_release, error=False):
@@ -509,6 +516,62 @@ async def run_and_check(spec, expected_release, error=False):
         )
 
 
+@pytest.mark.parametrize(
+    "rebase_in_current_stream",
+    [False, True],
+)
+@pytest.mark.asyncio
+async def test_update_release_maintenance_cs_branch(
+    rebase_in_current_stream,
+    minimal_spec,
+    autorelease_spec,
+    release_macro_spec,
+):
+    """Test that CentOS Stream branches for maintenance RHEL get Z-stream release bumping."""
+    package = "test"
+    dist_git_branch = "c8s"
+
+    async def mock_get_latest_candidate_build(package, dist_git_branch_arg):
+        return EVR(version="0.1", release="5.elX"), "dummy_ref"
+
+    flexmock(specfile_tools).should_receive("get_latest_candidate_build").replace_with(
+        mock_get_latest_candidate_build
+    )
+
+    async def mock_get_maintenance_rhel_branch(_):
+        return "rhel-8.10.0"
+
+    flexmock(specfile_tools).should_receive("get_maintenance_rhel_branch").replace_with(
+        mock_get_maintenance_rhel_branch
+    )
+
+    tool = UpdateReleaseTool()
+
+    async def run_and_check(spec, expected_release):
+        output = await tool.run(
+            input=UpdateReleaseToolInput(
+                spec=spec,
+                package=package,
+                dist_git_branch=dist_git_branch,
+                rebase=rebase_in_current_stream,
+            )
+        ).middleware(GlobalTrajectoryMiddleware(pretty=True))
+        result = output.result
+        assert result.startswith("Successfully")
+        release_line = next(line for line in spec.read_text().splitlines() if line.startswith("Release:"))
+        assert release_line == f"Release:        {expected_release}"
+
+    await run_and_check(minimal_spec, "0%{?dist}.1" if rebase_in_current_stream else "5%{?dist}.1")
+    await run_and_check(
+        autorelease_spec,
+        "0%{?dist}.%{autorelease -n}" if rebase_in_current_stream else "5%{?dist}.%{autorelease -n}",
+    )
+    await run_and_check(
+        release_macro_spec,
+        "0%{?dist}.1" if rebase_in_current_stream else "5%{?dist}.1",
+    )
+
+
 @pytest.mark.parametrize(
     "rebase",
     [False, True],
@@ -613,6 +676,13 @@ async def test_update_release_abandon_autorelease_non_zstream(autorelease_spec):
     package = "test"
     dist_git_branch = "c10s"
 
+    async def mock_get_maintenance_rhel_branch(_):
+        return None
+
+    flexmock(specfile_tools).should_receive("get_maintenance_rhel_branch").replace_with(
+        mock_get_maintenance_rhel_branch
+    )
+
     tool = UpdateReleaseTool()
 
     output = await tool.run(