Improve release setting on Y-Stream branches

Signed-off-by: Nikola Forró <nforro@redhat.com>

Author: nforro

agents/tests/unit/test_tools.py

@@ -258,6 +258,15 @@ async def run_and_check(spec, expected_release, error=False):
         await run_and_check(minimal_spec, "0%{?dist}.1" if rebase else "2%{?dist}.2")
         await run_and_check(autorelease_spec, "0%{?dist}.%{autorelease -n}" if rebase else "2%{?dist}.%{autorelease -n}")
 
+    with specfile.Specfile(minimal_spec) as spec:
+        spec.raw_release = "2%{?dist}.1"
+    with specfile.Specfile(autorelease_spec) as spec:
+        spec.raw_release = "2%{?dist}.%{autorelease -n}"
+
+    if not dist_git_branch.startswith("rhel-"):
+        await run_and_check(minimal_spec, "1%{?dist}" if rebase else "3%{?dist}")
+        await run_and_check(autorelease_spec, "%autorelease")
+
 
 @pytest.mark.asyncio
 async def test_get_cwd(tmp_path):

agents/tools/specfile.py

@@ -7,7 +7,7 @@
 from pydantic import BaseModel, Field
 from specfile import Specfile
 from specfile.utils import EVR
-from specfile.value_parser import EnclosedMacroSubstitution, MacroSubstitution, ValueParser
+from specfile.value_parser import EnclosedMacroSubstitution, MacroSubstitution, Node, ValueParser
 
 from beeai_framework.context import RunContext
 from beeai_framework.emitter import Emitter
@@ -165,12 +165,48 @@ async def _get_latest_higher_stream_build(package: str, candidate_tag: str) -> E
         return EVR(epoch=build["epoch"] or 0, version=build["version"], release=build["release"])
 
     @staticmethod
-    async def _bump_or_reset_release(spec_path: Path, rebase: bool) -> None:
+    def _parse_release(release: str) -> list[Node]:
+        return list(ValueParser.flatten(ValueParser.parse(release)))
+
+    @staticmethod
+    def _find_macro(name: str, nodes: list[Node]) -> int | None:
+        for index, node in reversed(list(enumerate(nodes))):
+            if (
+                isinstance(node, (MacroSubstitution, EnclosedMacroSubstitution))
+                and node.name == name
+            ):
+                return index
+        return None
+
+    @classmethod
+    async def _bump_or_reset_release(cls, spec_path: Path, rebase: bool) -> None:
         with Specfile(spec_path) as spec:
-            if rebase and not spec.has_autorelease:
-                spec.release = "1"
+            current_release = spec.raw_release
+        nodes = cls._parse_release(current_release)
+
+        autorelease_index = cls._find_macro("autorelease", nodes)
+        dist_index = cls._find_macro("dist", nodes)
+        if autorelease_index is not None:
+            # revert to plain %autorelease
+            release = "%autorelease"
+        else:
+            if dist_index is None:
+                prefix = current_release
+                suffix = ""
+            else:
+                prefix = "".join(str(n) for n in nodes[: dist_index])
+                suffix = "".join(str(n) for n in nodes[dist_index + 1 :])
+            if m := re.match(r"^(\d+)(.*)$", prefix):
+                # increase or reset the main numeric part
+                release = str(1 if rebase else int(m.group(1)) + 1) + m.group(2)
             else:
-                spec.bump_release()
+                release = prefix + ".1"
+            release += "%{?dist}"
+            if not re.match(r"^\.\d+$", suffix):
+                release += suffix
+
+        with Specfile(spec_path) as spec:
+            spec.raw_release = release
 
     @classmethod
     async def _set_zstream_release(
@@ -184,19 +220,10 @@ async def _set_zstream_release(
         higher_stream_base_release, _ = latest_higher_stream_build.release.rsplit(".el", maxsplit=1)
         with Specfile(spec_path) as spec:
             current_release = spec.raw_release
-        nodes = list(ValueParser.flatten(ValueParser.parse(current_release)))
-
-        def find_macro(name):
-            for index, node in reversed(list(enumerate(nodes))):
-                if (
-                    isinstance(node, (MacroSubstitution, EnclosedMacroSubstitution))
-                    and node.name == name
-                ):
-                    return index
-            return None
+        nodes = cls._parse_release(current_release)
 
-        autorelease_index = find_macro("autorelease")
-        dist_index = find_macro("dist")
+        autorelease_index = cls._find_macro("autorelease", nodes)
+        dist_index = cls._find_macro("dist", nodes)
         if autorelease_index is not None:
             if rebase:
                 # %autorelease present, rebase, reset the release