git-importer/lib/obs_revision.py

import datetime
import logging
import re
import xml.etree.ElementTree as ET
from urllib.error import HTTPError


class OBSRevision:
    def __init__(self, obs, history, project, package):
        self.obs = obs
        self.history = history
        self.project = project
        self.package = package

        self.commit = None
        self.ignored = False

    def parse(self, xml):
        self.rev = int(xml.get("rev"))
        self.unexpanded_srcmd5 = xml.find("srcmd5").text
        # Replaced in check_expanded
        self.srcmd5 = self.unexpanded_srcmd5

        time = int(xml.find("time").text)
        self.time = datetime.datetime.fromtimestamp(time)

        userid = xml.find("user")
        if userid is not None:
            self.userid = userid.text
        else:
            self.userid = "unknown"

        comment = xml.find("comment")
        if comment is not None:
            self.comment = comment.text or ""
        else:
            self.comment = ""

        # Populated by check_link
        self.linkrev = None

        self.requestid = None
        requestid = xml.find("requestid")
        if requestid is not None:
            self.requestid = int(requestid.text)
        else:
            # Sometimes requestid is missing, but can be extracted
            # from "comment"
            matched = re.match(
                r"^Copy from .* based on submit request (\d+) from user .*$",
                self.comment,
            )
            if matched:
                self.requestid = int(matched.group(1))

        return self

    def __str__(self):
        return f"Rev {self.project}/{self.rev} Md5 {self.srcmd5} {self.time} {self.userid} {self.requestid}"

    def __repr__(self):
        return f"[{self.__str__()}]"

    def check_link(self):
        """Add 'linkrev' attribute into the revision. Returns False if the link is invalid"""
        try:
            root = self.obs._xml(
                f"source/{self.project}/{self.package}/_link", rev=self.srcmd5
            )
        except HTTPError as e:
            if e.code == 404:
                logging.debug("No _link for the revision")
                return True
            raise e
        except ET.ParseError:
            logging.error(
                f"_link can't be parsed [{self.project}/{self.package} rev={self.srcmd5}]"
            )
            return False

        target_project = root.get("project")
        rev = self.history.find_last_rev_after_time(target_project, self.time)
        if rev:
            logging.debug(f"Linkrev found: {rev}")
            self.linkrev = rev.srcmd5
        return True

    def check_expanded(self):
        # Even if it's not a link we still need to check the expanded
        # srcmd5 as it's possible used in submit requests
        if not self.check_link():
            return False

        # If there is a "linkrev", "rev" is ignored
        params = {"rev": self.srcmd5, "expand": "1"}
        if self.linkrev:
            params["linkrev"] = self.linkrev

        try:
            root = self.obs._xml(f"source/{self.project}/{self.package}", **params)
        except HTTPError as e:
            if e.code == 400:
                logging.error(
                    f"Package [{self.project}/{self.package} {params}] can't be expanded: {e}"
                )
                return False
            raise e

        self.srcmd5 = root.get("srcmd5")
        return True