feat(manifest): Add ability to create on_commit_msg policies per bran…

…ches

.gitignore

@@ -1,4 +1,5 @@
 .devbox
+.idea/
 .venv/
 __pycache__/
 dist/

src/releaser/cli/commands/analyze_manifest.py

@@ -1,4 +1,5 @@
 """Create a release manifest."""
+
 from __future__ import annotations
 
 import argparse

src/releaser/cli/commands/bake_manifest.py

@@ -1,4 +1,5 @@
 """Create a release manifest."""
+
 from __future__ import annotations
 
 import argparse

src/releaser/cli/commands/create_manifest.py

@@ -1,4 +1,5 @@
 """Create a release manifest."""
+
 from __future__ import annotations
 
 import argparse

src/releaser/cli/commands/upload_manifest.py

@@ -1,4 +1,5 @@
 """Create a release manifest."""
+
 from __future__ import annotations
 
 import argparse

src/releaser/hexagon/entities/strategy/__init__.py

@@ -1,9 +1,13 @@
 from __future__ import annotations
 
-from .application import Application, ApplicationReleaseStrategy
 from .image import Image
 from .policy import CommitMsgMatchPolicy, GitCommitShaTag, LiteralTag, VersionTag
-from .release_strategy import ReleaseStrategy
+from .release_strategy import (
+    Application,
+    ApplicationReleaseStrategy,
+    ReleaseStrategy,
+    Rule,
+)
 
 __all__ = [
     "Application",
@@ -14,4 +18,5 @@
     "VersionTag",
     "LiteralTag",
     "Image",
+    "Rule",
 ]

src/releaser/hexagon/entities/strategy/policy.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Any, Literal
 
 
@@ -93,18 +93,47 @@ def parse_dict(cls, data: dict[str, Any]) -> "CommitMsgMatchPolicy":
         return cls(
             match=_aslist(data["match"]),
             tags=[
-                VersionTag.parse_dict(tag)
-                if tag.get("type") == "version"
-                else GitCommitShaTag.parse_dict(tag)
-                if tag.get("type") == "git_commit_sha"
-                else LiteralTag.parse_dict(tag)
+                (
+                    VersionTag.parse_dict(tag)
+                    if tag.get("type") == "version"
+                    else (
+                        GitCommitShaTag.parse_dict(tag)
+                        if tag.get("type") == "git_commit_sha"
+                        else LiteralTag.parse_dict(tag)
+                    )
+                )
                 for tag in data["tags"]
             ],
             filter=data.get("filter"),
             depth=data.get("depth", 20),
         )
 
 
+@dataclass(frozen=True)
+class Rule:
+    """A rule for a release strategy."""
+
+    branches: list[str] = field(default_factory=list)
+    """The branches to match for the HEAD commit message."""
+
+    commit_msg: list[CommitMsgMatchPolicy] = field(default_factory=list)
+    """The tags to match for the HEAD commit message.
+    These are global policies that apply to all applications.
+    """
+
+    @classmethod
+    def parse_dict(cls, data: dict[str, Any]) -> Rule:
+        """Parse a rule from a dictionary."""
+
+        return cls(
+            branches=_aslist(data.get("branch")),
+            commit_msg=[
+                CommitMsgMatchPolicy.parse_dict(policy)
+                for policy in data.get("commit_msg", [])
+            ],
+        )
+
+
 def _aslist(any: Any) -> list[Any]:
     """Convert a value to a list."""
     if any is None:

src/releaser/hexagon/entities/strategy/release_strategy.py

@@ -3,8 +3,46 @@
 from dataclasses import dataclass, field
 from typing import Any
 
-from .application import Application, ApplicationReleaseStrategy
-from .policy import CommitMsgMatchPolicy
+from .image import Image
+from .policy import Rule
+
+
+@dataclass
+class Application:
+    """An application within a development repository."""
+
+    on: list[Rule] | None = None
+    """The tags to match for the HEAD commit."""
+
+    images: list[Image] | None = None
+    """The images produced by the application.
+
+    Images must be a list of OCI-compliant image repositories.
+    """
+
+    @classmethod
+    def parse_dict(cls, data: dict[str, Any]) -> "Application":
+        """Parse an application from a dictionary."""
+        on = [Rule.parse_dict(rule) for rule in data.get("on", {}).values()]
+        return cls(
+            on=on,
+            images=[_asimage(image) for image in data.get("images", [])],
+        )
+
+
+@dataclass
+class ApplicationReleaseStrategy:
+    name: str
+    """The name of the application."""
+
+    on: list[Rule]
+    """The tags to match for the HEAD commit."""
+
+    images: list[Image]
+    """The images produced by the application.
+
+    Images must be a list of OCI-compliant image repositories.
+    """
 
 
 @dataclass(frozen=True)
@@ -14,9 +52,9 @@ class ReleaseStrategy:
     applications: dict[str, Application]
     """The applications in the repository. Each application may define a custom release strategy."""
 
-    on_commit_msg: list[CommitMsgMatchPolicy] = field(default_factory=list)
-    """The tags to match for the HEAD commit message.
-    These are global policies that apply to all applications.
+    on: list[Rule] = field(default_factory=list)
+    """The rules to set tags.
+    These are global rules that apply to all applications.
     """
 
     @classmethod
@@ -27,10 +65,7 @@ def parse_dict(cls, data: dict[str, Any]) -> "ReleaseStrategy":
                 name: Application.parse_dict(application)
                 for name, application in data.get("applications", {}).items()
             },
-            on_commit_msg=[
-                CommitMsgMatchPolicy.parse_dict(policy)
-                for policy in data.get("on_commit_msg", [])
-            ],
+            on=[Rule.parse_dict(rule) for rule in data.get("on", {}).values()],
         )
 
     def get_release_strategy_for_application(
@@ -41,8 +76,7 @@ def get_release_strategy_for_application(
             if app_name == name:
                 return ApplicationReleaseStrategy(
                     name=app_name,
-                    match_commit_msg=_aslist(application.on_commit_msg)
-                    or self.on_commit_msg,
+                    on=_aslist(application.on) or self.on,
                     images=application.images or [],
                 )
         raise ValueError(f"Application {name} not found in release strategy")
@@ -55,3 +89,10 @@ def _aslist(any: Any) -> list[Any]:
     if not isinstance(any, list):
         return [any]
     return any  # type: ignore
+
+
+def _asimage(value: str | dict[str, Any]) -> Image:
+    """Convert a value to an image."""
+    if isinstance(value, str):
+        return Image(repository=value)
+    return Image.parse_dict(value)

src/releaser/hexagon/errors.py

@@ -1,4 +1,5 @@
 """Custom errors for releaser hexagon."""
+
 from __future__ import annotations
 
 

src/releaser/hexagon/ports/git_reader.py

@@ -1,4 +1,5 @@
 """This module defines the GitReader abstract base class."""
+
 from __future__ import annotations
 
 import abc
@@ -11,6 +12,11 @@ def is_dirty(self) -> bool:
         """Check if the repository has uncommitted changes."""
         raise NotImplementedError
 
+    @abc.abstractmethod
+    def read_current_branch(self) -> str:
+        """Read the name of the current branch."""
+        raise NotImplementedError
+
     @abc.abstractmethod
     def read_most_recent_commit_sha(self) -> str:
         """Read the SHA of the latest commit."""
@@ -33,3 +39,9 @@ def read_last_commit_message(self, depth: int, filter: str | None) -> str | None
                     return commit_msg
             return None
         return history[0]
+
+    def current_reference_matches(self, ref: list[str]) -> bool:
+        """Check if the current reference matches the given ref."""
+        if not ref:
+            return True
+        return self.read_current_branch() in ref

src/releaser/hexagon/ports/json_writer.py

@@ -1,4 +1,5 @@
 """This module defines the JsonWriter abstract base class."""
+
 from __future__ import annotations
 
 import abc

src/releaser/hexagon/ports/strategy_reader.py

@@ -1,4 +1,5 @@
 """This module defines the StrategyReader abstract base class."""
+
 from __future__ import annotations
 
 import abc

src/releaser/hexagon/ports/version_reader.py

@@ -1,4 +1,5 @@
 """This module defines the VersionReader abstract base class."""
+
 from __future__ import annotations
 
 import abc

src/releaser/hexagon/ports/webhook_client.py

@@ -1,4 +1,5 @@
 """This module defines the HttpClient abstract base class."""
+
 from __future__ import annotations
 
 import abc

src/releaser/hexagon/services/manifest_baker.py

@@ -1,6 +1,5 @@
 """Service used to build manfest artefacts."""
 
-
 from dataclasses import dataclass
 
 from ..entities import bakery, strategy

src/releaser/hexagon/services/manifest_generator.py

@@ -1,4 +1,5 @@
 """Service used to generate manifest during a release."""
+
 from __future__ import annotations
 
 import re
@@ -75,13 +76,18 @@ def _generate_application_tags(
         self, application: strategy.ApplicationReleaseStrategy
     ) -> Iterator[str]:
         """Generate the list of tags to release for an application."""
-        for policy in application.match_commit_msg:
-            msg = self.git_reader.read_last_commit_message(policy.depth, policy.filter)
-            if not msg:
+        for rule in application.on:
+            if not self.git_reader.current_reference_matches(rule.branches):
                 continue
-            if self._verify_commit_msg_match_against_policy(msg, policy):
-                for tag in policy.tags:
-                    yield self._get_value_for_tag(tag)
+            for policy in rule.commit_msg:
+                msg = self.git_reader.read_last_commit_message(
+                    policy.depth, policy.filter
+                )
+                if not msg:
+                    continue
+                if self._verify_commit_msg_match_against_policy(msg, policy):
+                    for tag in policy.tags:
+                        yield self._get_value_for_tag(tag)
 
     def _verify_commit_msg_match_against_policy(
         self, commit_msg: str, policy: CommitMsgMatchPolicy

src/releaser/hexagon/services/manifest_notifier.py

@@ -1,5 +1,6 @@
 """Service used to send a POST request to a webhook URL with a
 manifest as JSON payload."""
+
 from __future__ import annotations
 
 from dataclasses import dataclass

src/releaser/infra/git_reader/stub.py

@@ -10,6 +10,12 @@ def __init__(self) -> None:
         self._sha: str | None = None
         self._history: list[str] | None = None
         self._is_dirty: bool | None = None
+        self._branch: str | None = None
+
+    def read_current_branch(self) -> str:
+        if self._branch is None:
+            raise RuntimeError("branch not set in stub git reader")
+        return self._branch
 
     def is_dirty(self) -> bool:
         if self._is_dirty is None:
@@ -37,3 +43,7 @@ def set_history(self, history: list[str]) -> None:
     def set_is_dirty(self, is_dirty: bool) -> None:
         """Test helper: Set the value that will be returned by is_dirty."""
         self._is_dirty = is_dirty
+
+    def set_branch(self, branch: str) -> None:
+        """ Test helper : set the value that be returned by read_current_branch."""
+        self._branch = branch