refactor: move static methods to free functions

Motivation: Adhere to project guidelines to prioritize functional style
over Object-Oriented and address reviewer feedback.

Design Choices:
- Extracted logic from @staticmethod methods into module-level functions.
- Kept original method names as minimal wrappers where necessary for
  backward compatibility and test stability.
- Updated tests to use module-level patching instead of patch.object where
  appropriate.

User Benefits:
- Improved code maintainability through explicit data flow.
- Enhanced testability by allowing logic to be verified in isolation.
- Cleaner class definitions focused on state management.

Author: okurz

openqabot/commenter.py

@@ -16,7 +16,7 @@
 
 from .loader.qem import get_aggregate_results, get_submission_results, get_submissions
 from .openqa import OpenQAInterface
-from .osclib.comments import CommentAPI
+from .osclib.comments import CommentAPI, add_marker, truncate
 
 if TYPE_CHECKING:
     from argparse import Namespace
@@ -26,6 +26,27 @@
 log = getLogger("bot.commenter")
 
 
+def _format_group_message(group_data: dict[str, Any]) -> str:
+    """Format a single group summary into a markdown string."""
+    msg = "\n\n" + group_data["title"]
+    infos = []
+    if group_data["passed"]:
+        infos.append(f"{group_data['passed']:d} tests passed")
+    if group_data["failed"]:
+        infos.append(f"{len(group_data['failed']):d} tests failed")
+    if group_data["unfinished"]:
+        infos.append(f"{group_data['unfinished']:d} unfinished tests")
+    msg += "(" + ", ".join(infos) + ")\n"
+    for fail in group_data["failed"]:
+        msg += fail
+    return msg
+
+
+def _escape_for_markdown(string: str) -> str:
+    """Escape underscores for markdown."""
+    return string.replace("_", r"\_")
+
+
 class Commenter:
     """Logic for commenting on submissions in OBS."""
 
@@ -87,8 +108,8 @@ def osc_comment(self, sub: Submission, msg: str, state: str) -> None:
             for key in sub.revisions:
                 info[f"revision_{key.version}_{key.arch}"] = sub.revisions[key]
 
-        msg = self.commentapi.add_marker(msg, bot_name, info)
-        msg = self.commentapi.truncate(msg.strip())
+        msg = add_marker(msg, bot_name, info)
+        msg = truncate(msg.strip())
 
         kw = {"request_id": str(sub.rr)}
         comments = self.commentapi.get_comments(**kw)
@@ -125,7 +146,7 @@ def summarize_message(self, jobs: list[dict[str, Any]]) -> str:
 
         msg = ""
         for group in sorted(groups.keys()):
-            msg += self._format_group_message(groups[group])
+            msg += _format_group_message(groups[group])
         return msg.rstrip("\n")
 
     def _process_job(self, groups: dict[str, dict[str, Any]], job: dict[str, Any]) -> None:
@@ -135,7 +156,7 @@ def _process_job(self, groups: dict[str, dict[str, Any]], job: dict[str, Any]) -
             log.warning("Job %s skipped: Missing 'job_group'", job["job_id"])
             return
 
-        gl = f"{Commenter.escape_for_markdown(job['job_group'])}@{Commenter.escape_for_markdown(job['flavor'])}"
+        gl = f"{_escape_for_markdown(job['job_group'])}@{_escape_for_markdown(job['flavor'])}"
         self._create_group_if_missing(groups, job, gl)
 
         job_summary = self.__summarize_one_openqa_job(job)
@@ -172,27 +193,6 @@ def _create_group_if_missing(self, groups: dict[str, dict[str, Any]], job: dict[
                 "failed": [],
             }
 
-    @staticmethod
-    def _format_group_message(group_data: dict[str, Any]) -> str:
-        """Format a single group summary into a markdown string."""
-        msg = "\n\n" + group_data["title"]
-        infos = []
-        if group_data["passed"]:
-            infos.append(f"{group_data['passed']:d} tests passed")
-        if group_data["failed"]:
-            infos.append(f"{len(group_data['failed']):d} tests failed")
-        if group_data["unfinished"]:
-            infos.append(f"{group_data['unfinished']:d} unfinished tests")
-        msg += "(" + ", ".join(infos) + ")\n"
-        for fail in group_data["failed"]:
-            msg += fail
-        return msg
-
-    @staticmethod
-    def escape_for_markdown(string: str) -> str:
-        """Escape underscores for markdown."""
-        return string.replace("_", r"\_")
-
     def __summarize_one_openqa_job(self, job: dict[str, Any]) -> str | None:
         testurl = osc.core.makeurl(self.client.openqa.baseurl, ["tests", str(job["job_id"])])
         name = job["name"]

openqabot/osclib/comments.py

@@ -30,6 +30,65 @@ def comment_as_dict(comment_element: etree.Element) -> dict[str, Any]:
     }
 
 
+def _parse_info(info_str: str) -> dict[str, str]:
+    """Parse the info string from a comment marker."""
+    if not (stripped := info_str.strip()):
+        return {}
+    return dict(pair.split("=") for pair in stripped.split())
+
+
+def _info_matches(info: dict[str, str], info_match: dict[str, Any] | None) -> bool:
+    """Check if parsed info matches the criteria."""
+    if not info_match:
+        return True
+    return all(value is None or info.get(key) == value for key, value in info_match.items())
+
+
+def add_marker(comment: str, bot: str, info: dict[str, Any] | None = None) -> str:
+    """Add bot marker to comment that can be used to find comment."""
+    info_str = ""
+    if info:
+        infos = ["=".join((str(key), str(value))) for key, value in info.items()]
+        info_str = " " + " ".join(infos)
+
+    marker = f"<!-- {bot}{info_str} -->"
+    return marker + "\n\n" + comment
+
+
+def truncate(comment: str, suffix: str = "...", length: int = 65535) -> str:
+    """Truncate a comment to a specific length, preserving markdown pre tags."""
+    # Handle very short length by dropping suffix and just chopping comment.
+    if length <= len(suffix) + len("\n</pre>"):
+        return comment[:length]
+    if len(comment) <= length:
+        return comment
+
+    # Determine the point at which to end by leaving room for suffix.
+    end = length - len(suffix)
+    if comment.find("<pre>", 0, end) != -1:
+        # For the sake of simplicity leave space for closing pre tag even if
+        # after truncation it may no longer be necessary. Otherwise, it
+        # requires recursion with some fun edge cases.
+        end -= len("\n</pre>")
+
+    # Check for the end location landing inside a pre tag and correct by
+    # moving in front of the tag. Landing on the ends is a noop.
+    pre_index = max(
+        comment.rfind("<pre>", end - 4, end + 4),
+        comment.rfind("</pre>", end - 5, end + 5),
+    )
+    if pre_index != -1:
+        end = pre_index
+
+    comment = comment[:end]
+
+    # Check for unbalanced pre tag and add a closing tag.
+    if comment.count("<pre>") > comment.count("</pre>"):
+        suffix += "\n</pre>"
+
+    return comment + suffix
+
+
 class OscCommentsValueError(ValueError):
     """Raised when an invalid value is provided to OscComments."""
 
@@ -98,20 +157,6 @@ def get_comments(
             comments[c["id"]] = c
         return comments
 
-    @staticmethod
-    def _parse_info(info_str: str) -> dict[str, str]:
-        """Parse the info string from a comment marker."""
-        if not (stripped := info_str.strip()):
-            return {}
-        return dict(pair.split("=") for pair in stripped.split())
-
-    @staticmethod
-    def _info_matches(info: dict[str, str], info_match: dict[str, Any] | None) -> bool:
-        """Check if parsed info matches the criteria."""
-        if not info_match:
-            return True
-        return all(value is None or info.get(key) == value for key, value in info_match.items())
-
     def comment_find(
         self,
         comments: dict[str, Any],
@@ -127,23 +172,12 @@ def comment_find(
             if bot != m.group("bot").lower():
                 continue
 
-            info = self._parse_info(m.group("info"))
-            if self._info_matches(info, info_match):
+            info = _parse_info(m.group("info"))
+            if _info_matches(info, info_match):
                 return c, info
 
         return None, None
 
-    @staticmethod
-    def add_marker(comment: str, bot: str, info: dict[str, Any] | None = None) -> str:
-        """Add bot marker to comment that can be used to find comment."""
-        info_str = ""
-        if info:
-            infos = ["=".join((str(key), str(value))) for key, value in info.items()]
-            info_str = " " + " ".join(infos)
-
-        marker = f"<!-- {bot}{info_str} -->"
-        return marker + "\n\n" + comment
-
     def add_comment(
         self,
         request_id: str | int | None = None,
@@ -163,46 +197,12 @@ def add_comment(
         if not comment:
             raise OscCommentsEmptyError
 
-        comment = self.truncate(comment.strip())
+        comment = truncate(comment.strip())
 
         query = {"parent_id": parent_id} if parent_id else {}
         url = self.prepare_url(request_id, project_name, package_name, query)
         return http_POST(url, data=comment)
 
-    @staticmethod
-    def truncate(comment: str, suffix: str = "...", length: int = 65535) -> str:
-        """Truncate a comment to a specific length, preserving markdown pre tags."""
-        # Handle very short length by dropping suffix and just chopping comment.
-        if length <= len(suffix) + len("\n</pre>"):
-            return comment[:length]
-        if len(comment) <= length:
-            return comment
-
-        # Determine the point at which to end by leaving room for suffix.
-        end = length - len(suffix)
-        if comment.find("<pre>", 0, end) != -1:
-            # For the sake of simplicity leave space for closing pre tag even if
-            # after truncation it may no longer be necessary. Otherwise, it
-            # requires recursion with some fun edge cases.
-            end -= len("\n</pre>")
-
-        # Check for the end location landing inside a pre tag and correct by
-        # moving in front of the tag. Landing on the ends is a noop.
-        pre_index = max(
-            comment.rfind("<pre>", end - 4, end + 4),
-            comment.rfind("</pre>", end - 5, end + 5),
-        )
-        if pre_index != -1:
-            end = pre_index
-
-        comment = comment[:end]
-
-        # Check for unbalanced pre tag and add a closing tag.
-        if comment.count("<pre>") > comment.count("</pre>"):
-            suffix += "\n</pre>"
-
-        return comment + suffix
-
     def delete(self, comment_id: str | int) -> None:
         """Remove a comment object.
 

openqabot/types/aggregate.py

@@ -30,6 +30,25 @@
 ALL_ISSUES_KEY = "TEST_ISSUES[]"
 
 
+def _normalize_repos(config: dict[str, Any]) -> dict[str, ProdVer]:
+    """Normalize repository configuration from config.settings."""
+    try:
+        return {key: ProdVer(*value.split(":")) for key, value in config["test_issues"].items()}
+    except KeyError as e:
+        raise NoTestIssuesError from e
+
+
+def get_buildnr(repohash: str, old_repohash: str, build: str) -> str:
+    """Determine the next build number based on current date and repohash."""
+    today = datetime.datetime.now(tz=UTC).date().strftime("%Y%m%d")
+
+    if build.startswith(today) and repohash == old_repohash:
+        raise SameBuildExistsError
+
+    counter = int(build.rsplit("-", maxsplit=1)[-1]) + 1 if build.startswith(today) else 1
+    return f"{today}-{counter}"
+
+
 class PostData(NamedTuple):
     """Data to be posted to dashboard."""
 
@@ -53,26 +72,12 @@ def __init__(self, config: JobConfig) -> None:
     @staticmethod
     def normalize_repos(config: dict[str, Any]) -> dict[str, ProdVer]:
         """Normalize repository configuration from config.settings."""
-        try:
-            return {key: ProdVer(*value.split(":")) for key, value in config["test_issues"].items()}
-        except KeyError as e:
-            raise NoTestIssuesError from e
+        return _normalize_repos(config)
 
     def __repr__(self) -> str:
         """Return a string representation of the Aggregate."""
         return f"<Aggregate product: {self.product}>"
 
-    @staticmethod
-    def get_buildnr(repohash: str, old_repohash: str, build: str) -> str:
-        """Determine the next build number based on current date and repohash."""
-        today = datetime.datetime.now(tz=UTC).date().strftime("%Y%m%d")
-
-        if build.startswith(today) and repohash == old_repohash:
-            raise SameBuildExistsError
-
-        counter = int(build.rsplit("-", maxsplit=1)[-1]) + 1 if build.startswith(today) else 1
-        return f"{today}-{counter}"
-
     def filter_submissions(self, submissions: list[Submission]) -> list[Submission]:
         """Filter out submissions that are not suitable for aggregate tests."""
 
@@ -230,7 +235,7 @@ def process_arch(
         old_build = old_jobs[0].get("build", "") if old_jobs else ""
 
         try:
-            build = self.get_buildnr(
+            build = get_buildnr(
                 repohash,
                 old_repohash,
                 old_build,