Feat: Add utility function for getting previous errata

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,11 +1,11 @@
 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
 from typing_extensions import Literal
+import re
 
 from bs4 import BeautifulSoup, Tag  # type: ignore
 from pydantic import BaseModel
@@ -29,11 +29,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 +61,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: ...
 
@@ -89,9 +96,14 @@ def get_erratum(erratum_id: str | int, full: bool = False) -> Erratum | FullErra
         for jira_issue_data in 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"],
@@ -100,6 +112,8 @@ def get_erratum(erratum_id: str | int, full: bool = False) -> Erratum | FullErra
         synopsis=details["synopsis"],
         status=ErrataStatus(details["status"]),
         all_issues_release_pending=all_issues_release_pending,
+        group_id=details["group_id"],
+        publish_date=publish_date,
         last_status_transition_timestamp=last_status_transition_timestamp,
     )
 
@@ -145,6 +159,146 @@ def get_erratum_for_link(link: str, full: bool = True) -> Erratum | FullErratum:
     return get_erratum(erratum_id, full=full)
 
 
+class RHELStream(StrEnum):
+    Z = "Z"
+    GA = "GA"
+
+
+class RHELVersion(BaseModel):
+    major_version: int
+    minor_version: int
+    stream: RHELStream
+
+    def __str__(self):
+        match self.major_version:
+            case 10:
+                return f"RHEL-{self.major_version}.{self.minor_version}.{self.stream}"
+            case 9 | 8:
+                if self.stream == RHELStream.Z:
+                    if self.minor_version % 2 == 1:
+                        return (
+                            f"RHEL-{self.major_version}.{self.minor_version}.0.Z.MAIN"
+                        )
+                    else:
+                        return f"RHEL-{self.major_version}.{self.minor_version}.0.Z.MAIN+EUS"
+                return f"RHEL-{self.major_version}.{self.minor_version}.0.GA"
+            case _:
+                return NotImplemented
+
+    def __sub__(self, other: int):
+        if not isinstance(other, int):
+            return NotImplemented
+
+        if self.stream == RHELStream.Z:
+            return RHELVersion(
+                major_version=self.major_version,
+                minor_version=self.minor_version,
+                stream=RHELStream("GA"),
+            )
+
+        return RHELVersion(
+            major_version=self.major_version,
+            minor_version=self.minor_version - 1,
+            stream=RHELStream("Z"),
+        )
+
+
+class RHELRelease(BaseModel):
+    version: str
+    # None means already shipped
+    ship_date: datetime | None
+
+
+def get_RHEL_version(version_string: str):
+    pattern = r"RHEL-(\d+)\.(\d+)(\.\d+)?\.(Z|GA)"
+    match = re.match(pattern, version_string)
+    if match is not None:
+        return RHELVersion(
+            major_version=int(match.group(1)),
+            minor_version=int(match.group(2)),
+            stream=RHELStream(match.group(4)),
+        )
+
+
+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 = (
+        datetime.strptime(ship_date_string, "%Y-%m-%dT%H:%M:%SZ").replace(
+            tzinfo=timezone.utc
+        )
+        if ship_date_string
+        else None
+    )
+
+    return RHELRelease(version=release_data["attributes"]["name"], ship_date=ship_date)
+
+
+def get_previous_erratum(current_erratum_id: str | int, package_name: str):
+    errata = get_erratum(current_erratum_id)
+
+    target_release = get_RHEL_release(errata.group_id)
+    cur_version = get_RHEL_version(target_release.version)
+    if cur_version is None:
+        logger.info(f"Unknow RHEL release format: {target_release.version}")
+        return None
+
+    related_erratum = ET_api_get(f"packages/{package_name}")["data"]["relationships"][
+        "errata"
+    ]
+    assert isinstance(related_erratum, list)
+
+    rel_prep_lookup = dict()
+
+    for errata in related_erratum:
+        if errata["status"] != ErrataStatus.REL_PREP:
+            continue
+
+        id = errata["id"]
+        cur_errata = get_erratum(id)
+        cur_release = get_RHEL_release(cur_errata.group_id)
+
+        # If the target release of REL_PREP is same as current target release
+        # Pick it without checking publish date
+        if cur_release.version == target_release.version:
+            logger.info("Exactly same target release from REL_PREP")  # testing purpose
+            return get_erratum(id)
+
+        rel_prep_lookup[cur_release.version] = {
+            "id": id,
+            "publish_date": cur_errata.publish_date,
+        }
+
+    cur_time = datetime.now(tz=timezone.utc)
+    while cur_version.minor_version >= 0:
+        # No need to check REL_PREP if the target release has been released
+        if target_release.ship_date is not None and str(cur_version) in rel_prep_lookup:
+            rel_prep = rel_prep_lookup[str(cur_version)]
+            if rel_prep["publish_date"] is None:
+                logger.info(
+                    f"Encounter REL_PREP errata without publish_date -> Errata ID: {rel_prep['id']}"
+                )
+            elif rel_prep["publish_date"] <= target_release.ship_date:
+                logger.info("From REL_PREP")  # testing purpose
+                return get_erratum(rel_prep["id"])
+
+        release = get_RHEL_release(str(cur_version))
+        if release.ship_date is None or release.ship_date <= cur_time:
+            released_build = ET_api_get(
+                f"product_versions/{release.version}/released_builds/{package_name}"
+            )
+            logger.info("From released build")  # testing purpose
+            return get_erratum(released_build["errata_id"])
+
+        cur_version -= 1
+
+
 class RuleParseError(Exception):
     pass
 

supervisor/supervisor_types.py

@@ -54,6 +54,8 @@ class Erratum(BaseModel):
     synopsis: str
     status: ErrataStatus
     all_issues_release_pending: bool
+    group_id: int
+    publish_date: datetime | None
     last_status_transition_timestamp: datetime
 
 class FullErratum(Erratum):