Feat: Add utility function for getting previous errata (#241)

This ultility function finds the previous errata with given errata ID (current)
and package name. The output will be a errata with the closest RHEL release
version. Here we pick REL_PREP over SHIPPED_ALIVE.

Author: Jazzcort

supervisor/errata_utils.py

@@ -1,10 +1,12 @@
+from __future__ import annotations
+from collections import defaultdict
 from datetime import datetime, timezone
 from enum import StrEnum
-from datetime import datetime
 from functools import cache
 import logging
 import os
-from typing import overload
+import re
+from typing import DefaultDict, overload
 from typing_extensions import Literal
 
 from bs4 import BeautifulSoup, Tag  # type: ignore
@@ -29,11 +31,12 @@ def ET_verify() -> bool | str:
         return True
 
 
-def ET_api_get(path: str):
+def ET_api_get(path: str, *, params: dict | None = None):
     response = requests_session().get(
         f"{ET_URL}/api/v1/{path}",
         auth=HTTPSPNEGOAuth(opportunistic_auth=True),
         verify=ET_verify(),
+        params=params,
     )
     response.raise_for_status()
     return response.json()
@@ -60,6 +63,12 @@ def ET_get_html(path: str):
     return response.text
 
 
+def get_utc_timestamp_from_str(timestamp_string: str):
+    return datetime.strptime(timestamp_string, "%Y-%m-%dT%H:%M:%SZ").replace(
+        tzinfo=timezone.utc
+    )
+
+
 @overload
 def get_erratum(erratum_id: str | int, full: Literal[False] = False) -> Erratum: ...
 
@@ -84,9 +93,14 @@ def get_erratum(erratum_id: str | int, full: bool = False) -> Erratum | FullErra
 
     jira_issues = [i["jira_issue"]["key"] for i in data["jira_issues"]["jira_issues"]]
 
-    last_status_transition_timestamp = datetime.strptime(
-        details["status_updated_at"], "%Y-%m-%dT%H:%M:%SZ"
-    ).replace(tzinfo=timezone.utc)
+    last_status_transition_timestamp = get_utc_timestamp_from_str(
+        details["status_updated_at"]
+    )
+    publish_date = (
+        get_utc_timestamp_from_str(details["publish_date"])
+        if details["publish_date"] is not None
+        else None
+    )
 
     base_erratum = Erratum(
         id=details["id"],
@@ -95,6 +109,8 @@ def get_erratum(erratum_id: str | int, full: bool = False) -> Erratum | FullErra
         synopsis=details["synopsis"],
         status=ErrataStatus(details["status"]),
         jira_issues=jira_issues,
+        release_id=details["group_id"],
+        publish_date=publish_date,
         last_status_transition_timestamp=last_status_transition_timestamp,
     )
 
@@ -140,6 +156,205 @@ def get_erratum_for_link(link: str, full: bool = False) -> Erratum | FullErratum
     return get_erratum(erratum_id, full=full)
 
 
+class RHELVersion(BaseModel):
+    major: int
+    minor: int
+    micro: int | None
+    stream: str
+
+    def __str__(self):
+        if self.micro is not None:
+            return f"RHEL-{self.major}.{self.minor}.{self.micro}.{self.stream}"
+
+        return f"RHEL-{self.major}.{self.minor}.{self.stream}"
+
+    @property
+    def parent(self) -> RHELVersion | None:
+        """The release that the release inherits builds from."""
+
+        if self.stream != "GA":
+            return RHELVersion(
+                major=self.major,
+                minor=self.minor,
+                micro=self.micro,
+                stream="GA",
+            )
+
+        if self.minor > 0:
+            one_minor_version_up = self.minor - 1
+            match self.major:
+                case 10:
+                    return RHELVersion(
+                        major=self.major,
+                        minor=one_minor_version_up,
+                        micro=self.micro,
+                        stream="Z",
+                    )
+                case 9 | 8:
+                    if one_minor_version_up % 2 == 1:
+                        return RHELVersion(
+                            major=self.major,
+                            minor=one_minor_version_up,
+                            micro=self.micro,
+                            stream="Z.MAIN",
+                        )
+                    else:
+                        return RHELVersion(
+                            major=self.major,
+                            minor=one_minor_version_up,
+                            micro=self.micro,
+                            stream="Z.MAIN+EUS",
+                        )
+
+        return None
+
+    @staticmethod
+    def from_str(version_string: str) -> RHELVersion | None:
+        version_string = version_string.strip()
+        version_string = version_string.upper()
+        pattern = r"RHEL-(\d+)\.(\d+)(?:\.(\d+))?\.([^\d].*)$"
+        match = re.match(pattern, version_string)
+        if match is not None:
+            version = RHELVersion(
+                major=int(match.group(1)),
+                minor=int(match.group(2)),
+                micro=int(match.group(3)) if match.group(3) else None,
+                stream=match.group(4),
+            )
+
+            assert version_string == str(version)
+
+            return version
+
+
+class RHELRelease(BaseModel):
+    version: str
+    # None means already shipped
+    ship_date: datetime | None
+
+    @property
+    def shipped(self):
+        return self.ship_date is None or self.ship_date < datetime.now(tz=timezone.utc)
+
+
+def get_RHEL_release(param: int | str):
+    response = (
+        ET_api_get("releases", params={"filter[id]": param})
+        if isinstance(param, int)
+        else ET_api_get("releases", params={"filter[name]": param})
+    )
+    release_data = response["data"][0]
+
+    ship_date_string = release_data["attributes"]["ship_date"]
+    ship_date = (
+        get_utc_timestamp_from_str(ship_date_string)
+        if ship_date_string is not None
+        else None
+    )
+
+    return RHELRelease(
+        version=release_data["attributes"]["name"],
+        ship_date=ship_date,
+    )
+
+
+def _get_rel_prep_lookup(package_name: str) -> DefaultDict[str, list[Erratum]]:
+    """Builds a lookup of REL_PREP errata for a package, keyed by RHEL release version.
+
+    This function queries an API for all errata associated with a given package,
+    filters for those in the "REL_PREP" (Release Preparation) state, and organizes
+    them into a dictionary where each key is a RHEL version string.
+
+    Args:
+        package_name: The name of the package to look up.
+
+    Returns:
+        A defaultdict where keys are RHEL release version strings and values are
+        lists of associated Erratum objects in the REL_PREP state.
+    """
+    rel_prep_lookup: DefaultDict[str, list[Erratum]] = defaultdict(list)
+    related_errata = ET_api_get(f"packages/{package_name}")["data"]["relationships"][
+        "errata"
+    ]
+    assert isinstance(related_errata, list)
+    for erratum_info in related_errata:
+        if erratum_info["status"] != ErrataStatus.REL_PREP:
+            continue
+
+        id = erratum_info["id"]
+        cur_erratum = get_erratum(id)
+        cur_release = get_RHEL_release(cur_erratum.release_id)
+
+        rel_prep_lookup[cur_release.version].append(cur_erratum)
+
+    return rel_prep_lookup
+
+
+def get_previous_erratum(current_erratum_id: str | int, package_name: str):
+    """Finds the previous erratum for a given package, starting from a specific erratum.
+
+    RHEL releases inherit packages from previous releases, but only until they are shipped.
+    This function searches backwards through RHEL release versions starting from the one
+    associated with current_erratum_id looking for applicable REL_PREP errata, or for
+    a shipped release. If we find a shipped release, we're done - no errata will be inherited
+    from previous releases, so if we haven't found a REL_PREP errata first, we use the
+    errata associated with the official released build for the package in that release
+    and stop.
+
+    Args:
+        current_erratum_id: The ID of the erratum to start the search from.
+        package_name: The name of the package for which to find the previous erratum.
+
+    Returns:
+        The previous Erratum object, or None if one cannot be found.
+    """
+    erratum = get_erratum(current_erratum_id)
+
+    target_release = get_RHEL_release(erratum.release_id)
+    target_version = RHELVersion.from_str(target_release.version)
+    if target_version is None:
+        logger.info(f"Unknown RHEL release format: {target_release.version}")
+        return None
+
+    def is_previous_erratum_applicable(erratum_version: str, erratum: Erratum):
+        if erratum_version == target_version:
+            return True
+        elif target_release.shipped:
+            return False
+
+        assert target_release.ship_date is not None
+        return (
+            erratum.publish_date is not None
+            and erratum.publish_date <= target_release.ship_date
+        )
+
+    rel_prep_lookup = _get_rel_prep_lookup(package_name)
+    cur_version = target_version
+    while cur_version:
+        rel_prep_errata = rel_prep_lookup[str(cur_version)]
+        rel_prep = [
+            e
+            for e in rel_prep_errata
+            if is_previous_erratum_applicable(str(cur_version), e)
+        ]
+
+        if rel_prep:
+            latest_erratum = max(
+                rel_prep,
+                key=lambda e: e.publish_date if e.publish_date else datetime.min,
+            )
+            return latest_erratum
+
+        release = get_RHEL_release(str(cur_version))
+        if release.shipped:
+            released_build = ET_api_get(
+                f"product_versions/{release.version}/released_builds/{package_name}"
+            )
+            return get_erratum(released_build["errata_id"])
+
+        cur_version = cur_version.parent
+
+
 class RuleParseError(Exception):
     pass
 

supervisor/supervisor_types.py

@@ -55,6 +55,8 @@ class Erratum(BaseModel):
     synopsis: str
     status: ErrataStatus
     jira_issues: list[str]
+    release_id: int
+    publish_date: datetime | None
     last_status_transition_timestamp: datetime