forked from importers/git-importer
118 lines
3.5 KiB
Python
118 lines
3.5 KiB
Python
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 read_link(self):
|
|
try:
|
|
return self.obs._xml(
|
|
f"source/{self.project}/{self.package}/_link",
|
|
rev=self.unexpanded_srcmd5,
|
|
)
|
|
except HTTPError as e:
|
|
if e.code == 404:
|
|
logging.debug("No _link for the revision")
|
|
return None
|
|
raise e
|
|
|
|
def check_link(self):
|
|
"""Add 'linkrev' attribute into the revision. Returns False if the link is invalid"""
|
|
try:
|
|
root = self.read_link()
|
|
if root is None:
|
|
return True
|
|
target_project = root.get("project")
|
|
except ET.ParseError:
|
|
logging.error(
|
|
f"_link can't be parsed [{self.project}/{self.package} rev={self.unexpanded_srcmd5}]"
|
|
)
|
|
return False
|
|
|
|
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
|