1
0
mirror of https://github.com/openSUSE/osc.git synced 2025-01-29 15:56:17 +01:00

Add obs_api.Request class for handling requests

This commit is contained in:
Daniel Mach 2024-03-01 14:09:07 +01:00
parent e5370b9c0b
commit 9078bc257c
23 changed files with 798 additions and 0 deletions

View File

@ -1,3 +1,4 @@
from .package import Package
from .package_sources import PackageSources
from .project import Project
from .request import Request

View File

@ -67,6 +67,13 @@ class LocalRole(str, Enum):
READER = "reader"
class ObsRatings(str, Enum):
LOW = "low"
MODERATE = "moderate"
IMPORTANT = "important"
CRITICAL = "critical"
class RebuildModes(str, Enum):
TRANSITIVE = "transitive"
DIRECT = "direct"
@ -77,3 +84,13 @@ class ReleaseTriggers(str, Enum):
MANUAL = "manual"
MAINTENANCE = "maintenance"
OBSGENDIFF = "obsgendiff"
class RequestStates(str, Enum):
REVIEW = "review"
NEW = "new"
ACCEPTED = "accepted"
DECLINED = "declined"
REVOKED = "revoked"
SUPERSEDED = "superseded"
DELETED = "deleted"

118
osc/obs_api/request.py Normal file
View File

@ -0,0 +1,118 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
from .enums import ObsRatings
from .request_action import RequestAction
from .request_history import RequestHistory
from .request_review import RequestReview
from .request_state import RequestState
class Request(XmlModel):
XML_TAG = "request"
id: Optional[str] = Field(
xml_attribute=True,
)
actions: Optional[int] = Field(
xml_attribute=True,
)
creator: Optional[str] = Field(
xml_attribute=True,
)
action_list: List[RequestAction] = Field(
xml_name="action",
)
state: Optional[RequestState] = Field(
)
description: Optional[str] = Field(
)
priority: Optional[ObsRatings] = Field(
)
review_list: Optional[List[RequestReview]] = Field(
xml_name="review",
)
history_list: Optional[List[RequestHistory]] = Field(
xml_name="history",
)
title: Optional[str] = Field(
)
accept_at: Optional[str] = Field(
)
@classmethod
def from_api(
cls,
apiurl: str,
request_id: int,
*,
with_history: Optional[bool] = None,
with_full_history: Optional[bool] = None
) -> "Request":
"""
Return the specified request.
:param request_id: Id of the request.
:param withhistory: Include the request history in the results.
:param withfullhistory: Includes both, request and review history in the results.
"""
url_path = ["request", request_id]
url_query = {
"withhistory": with_history,
"withfullhistory": with_full_history,
}
response = cls.xml_request("GET", apiurl, url_path, url_query)
return cls.from_file(response, apiurl=apiurl)
@classmethod
def cmd_diff(
cls,
apiurl: str,
request_id: int,
*,
with_issues: Optional[bool] = None,
with_description_issues: Optional[bool] = None,
diff_to_superseded: Optional[int] = None
) -> "Request":
"""
Return the specified request including a diff of all packages in the request.
:param request_id: Id of the request.
:param with_issues: Include parsed issues from referenced sources in the change files.
:param with_description_issues: Include parsed issues from request description.
:param diff_to_superseded: Diff relatively to the given superseded request.
"""
url_path = ["request", str(request_id)]
url_query = {
"cmd": "diff",
"view": "xml",
"withissues": with_issues,
"withdescriptionissues": with_description_issues,
"diff_to_superseded": diff_to_superseded,
}
response = cls.xml_request("POST", apiurl, url_path, url_query)
return cls.from_file(response, apiurl=apiurl)
def get_issues(self):
"""
Aggregate issues from action/sourcediff into a single list.
The list may contain duplicates.
To get any issues returned, it is crucial to load the request with the issues
by calling ``cmd_diff()`` with appropriate arguments first.
"""
result = []
for action in self.action_list or []:
if action.sourcediff is None:
continue
for issue in action.sourcediff.issue_list or []:
result.append(issue)
return result

View File

@ -0,0 +1,236 @@
import urllib.error
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
from .package import Package
from .package_sources import PackageSources
from .request_action_acceptinfo import RequestActionAcceptinfo
from .request_action_group import RequestActionGroup
from .request_action_grouped import RequestActionGrouped
from .request_action_options import RequestActionOptions
from .request_action_person import RequestActionPerson
from .request_action_source import RequestActionSource
from .request_action_target import RequestActionTarget
from .request_sourcediff import RequestSourcediff
class RequestAction(XmlModel):
XML_TAG = "action"
class TypeEnum(str, Enum):
SUBMIT = "submit"
DELETE = "delete"
CHANGE_DEVEL = "change_devel"
ADD_ROLE = "add_role"
SET_BUGOWNER = "set_bugowner"
MAINTENANCE_INCIDENT = "maintenance_incident"
MAINTENANCE_RELEASE = "maintenance_release"
RELEASE = "release"
GROUP = "group"
type: TypeEnum = Field(
xml_attribute=True,
)
source: Optional[RequestActionSource] = Field(
)
target: Optional[RequestActionTarget] = Field(
)
person: Optional[RequestActionPerson] = Field(
)
group: Optional[RequestActionGroup] = Field(
)
grouped_list: Optional[List[RequestActionGrouped]] = Field(
xml_name="grouped",
)
options: Optional[RequestActionOptions] = Field(
)
sourcediff: Optional[RequestSourcediff] = Field(
)
acceptinfo: Optional[RequestActionAcceptinfo] = Field(
)
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._allow_new_attributes = True
# source and target always come from ``self._apiurl`` while devel and factory projects may live elsewhere
self._devel_apiurl = self._apiurl
self._factory_apiurl = self._apiurl
self._factory_project = "openSUSE:Factory"
self._props = {}
self._allow_new_attributes = False
def _get_package(self, package_type):
key = f"{package_type}_package"
if key not in self._props:
func = getattr(self, f"_get_{package_type}_apiurl_project_package")
apiurl, project, package = func()
if apiurl is None:
self._props[key] = None
else:
try:
self._props[key] = Package.from_api(apiurl, project, package)
except urllib.error.HTTPError as e:
if e.code != 404:
raise
self._props[key] = None
return self._props[key]
def _get_package_sources(self, package_type, *, rev=None):
key = f"{package_type}_package_sources"
if key not in self._props:
func = getattr(self, f"_get_{package_type}_apiurl_project_package")
apiurl, project, package = func()
if apiurl is None:
self._props[key] = None
else:
try:
self._props[key] = PackageSources.from_api(apiurl, project, package, rev=rev)
except urllib.error.HTTPError as e:
if e.code != 404:
raise
self._props[key] = None
return self._props[key]
def _get_source_apiurl_project_package(self):
return self._apiurl, self.source.project, self.source.package
@property
def source_package(self) -> Optional[Package]:
"""
Return a ``Package`` object that encapsulates metadata of the source package.
"""
return self._get_package("source")
@property
def source_package_sources(self) -> Optional[PackageSources]:
"""
Return a ``PackageSources`` object that contains information about the ``source.rev`` revision of the source package sources in OBS SCM.
"""
if self.source is None:
return None
return self._get_package_sources("source", rev=self.source.rev)
def _get_target_apiurl_project_package(self):
if self.target is None:
return None, None, None
target_project, target_package = self.get_actual_target_project_package()
return self._apiurl, target_project, target_package
@property
def target_package(self) -> Optional[Package]:
"""
Return a ``Package`` object that encapsulates metadata of the target package.
"""
return self._get_package("target")
@property
def target_package_sources(self) -> Optional[PackageSources]:
"""
Return a ``PackageSources`` object that contains information about the current revision of the target package sources in OBS SCM.
"""
return self._get_package_sources("target")
def _get_factory_apiurl_project_package(self):
if self.target is None:
# a new package was submitted, it doesn't exist on target; let's read the package name from the source
target_project, target_package = None, self.source.package
else:
target_project, target_package = self.get_actual_target_project_package()
if (self._apiurl, target_project) == (self._factory_apiurl, self._factory_project):
# factory package equals the target package
return None, None, None
return self._factory_apiurl, self._factory_project, target_package
@property
def factory_package(self) -> Optional[Package]:
"""
Return a ``Package`` object that encapsulates metadata of the package in the factory project.
The name of the package equals the target package name.
"""
return self._get_package("factory")
@property
def factory_package_sources(self) -> Optional[PackageSources]:
"""
Return a ``PackageSources`` object that contains information about the current revision of the factory package sources in OBS SCM.
"""
return self._get_package_sources("factory")
def _get_devel_apiurl_project_package(self):
if self.factory_package is None:
return None, None, None
devel = self.factory_package.devel
if devel is None:
return None, None, None
return (
self._devel_apiurl,
devel.project,
devel.package or self.factory_package.name,
)
@property
def devel_package(self) -> Optional[Package]:
"""
Return a ``Package`` object that encapsulates metadata of the package in the devel project.
The devel project name and package name come from ``self.factory_package.devel``.
If the devel package name is not set, target package name is used.
"""
return self._get_package("devel")
@property
def devel_package_sources(self) -> Optional[PackageSources]:
"""
Return a ``PackageSources`` object that contains information about the current revision of the devel package sources in OBS SCM.
"""
return self._get_package_sources("devel")
def get_actual_target_project_package(self) -> Tuple[str, str]:
"""
Return the target project and package names because maintenance incidents require special handling.
The target project for maintenance incidents is virtual and cannot be queried.
The actual target project is specified in target's ``releaseproject`` field.
Also the target package for maintenance incidents is not set explicitly.
It is extracted from ``releasename`` field from the source metadata.
If ``releasename`` is not defined, source package name is used.
"""
if self.type == "maintenance_incident":
# dmach's note on security:
# The ``releaseproject`` is baked into the target information in the request and that's perfectly ok.
# The ``releasename`` is part of the source package metadata and *may change* after the request is created.
# After consulting this with OBS developers, I believe this doesn't represent any security issue
# because the project is fixed and tampering with ``releasename`` might only lead to inconsistent naming,
# the package would still end up it the same project.
# target.releaseproject is always set for a maintenance_incident
assert self.target
assert self.target.releaseproject
project = self.target.releaseproject
# the target package is not specified
# we need to extract it from source package's metadata or use source package name as a fallback
assert self.source_package
if self.source_package.releasename:
package = self.source_package.releasename.split(".")[0]
else:
package = self.source_package.name
return project, package
assert self.target
assert self.target.project
assert self.target.package
return self.target.project, self.target.package

View File

@ -0,0 +1,33 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
class RequestActionAcceptinfo(XmlModel):
XML_TAG = "acceptinfo"
rev: str = Field(
xml_attribute=True,
)
srcmd5: str = Field(
xml_attribute=True,
)
osrcmd5: str = Field(
xml_attribute=True,
)
oproject: Optional[str] = Field(
xml_attribute=True,
)
opackage: Optional[str] = Field(
xml_attribute=True,
)
xsrcmd5: Optional[str] = Field(
xml_attribute=True,
)
oxsrcmd5: Optional[str] = Field(
xml_attribute=True,
)

View File

@ -0,0 +1,13 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
class RequestActionGroup(XmlModel):
XML_TAG = "group"
name: str = Field(
xml_attribute=True,
)
role: Optional[str] = Field(
xml_attribute=True,
)

View File

@ -0,0 +1,9 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
class RequestActionGrouped(XmlModel):
XML_TAG = "grouped"
id: str = Field(
xml_attribute=True,
)

View File

@ -0,0 +1,27 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
class RequestActionOptions(XmlModel):
XML_TAG = "options"
class SourceupdateEnum(str, Enum):
UPDATE = "update"
NOUPDATE = "noupdate"
CLEANUP = "cleanup"
sourceupdate: Optional[SourceupdateEnum] = Field(
)
class UpdatelinkEnum(str, Enum):
TRUE = "true"
FALSE = "false"
updatelink: Optional[UpdatelinkEnum] = Field(
)
class MakeoriginolderEnum(str, Enum):
TRUE = "true"
FALSE = "false"
makeoriginolder: Optional[MakeoriginolderEnum] = Field(
)

View File

@ -0,0 +1,13 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
class RequestActionPerson(XmlModel):
XML_TAG = "person"
name: str = Field(
xml_attribute=True,
)
role: Optional[str] = Field(
xml_attribute=True,
)

View File

@ -0,0 +1,17 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
class RequestActionSource(XmlModel):
XML_TAG = "source"
project: str = Field(
xml_attribute=True,
)
package: Optional[str] = Field(
xml_attribute=True,
)
rev: Optional[str] = Field(
xml_attribute=True,
)

View File

@ -0,0 +1,21 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
class RequestActionTarget(XmlModel):
XML_TAG = "target"
project: str = Field(
xml_attribute=True,
)
package: Optional[str] = Field(
xml_attribute=True,
)
releaseproject: Optional[str] = Field(
xml_attribute=True,
)
repository: Optional[str] = Field(
xml_attribute=True,
)

View File

@ -0,0 +1,19 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
class RequestHistory(XmlModel):
XML_TAG = "history"
who: str = Field(
xml_attribute=True,
)
when: str = Field(
xml_attribute=True,
)
description: str = Field(
)
comment: Optional[str] = Field(
)

View File

@ -0,0 +1,57 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
from .enums import RequestStates
from .request_review_history import RequestReviewHistory
class RequestReview(XmlModel):
XML_TAG = "review"
state: RequestStates = Field(
xml_attribute=True,
)
created: Optional[str] = Field(
xml_attribute=True,
)
by_user: Optional[str] = Field(
xml_attribute=True,
)
by_group: Optional[str] = Field(
xml_attribute=True,
)
by_project: Optional[str] = Field(
xml_attribute=True,
)
by_package: Optional[str] = Field(
xml_attribute=True,
)
who: Optional[str] = Field(
xml_attribute=True,
)
when: Optional[str] = Field(
xml_attribute=True,
)
comment: Optional[str] = Field(
)
history_list: Optional[List[RequestReviewHistory]] = Field(
xml_name="history",
)
def get_user_and_type(self):
if self.by_user:
return (self.by_user, "user")
if self.by_group:
return (self.by_group, "group")
if self.by_package:
return (f"{self.by_project}/{self.by_package}", "package")
if self.by_project:
return (self.by_project, "project")
raise RuntimeError("Unable to determine user and its type")

View File

@ -0,0 +1,19 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
class RequestReviewHistory(XmlModel):
XML_TAG = "history"
who: str = Field(
xml_attribute=True,
)
when: str = Field(
xml_attribute=True,
)
description: str = Field(
)
comment: Optional[str] = Field(
)

View File

@ -0,0 +1,30 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
from .request_sourcediff_files_file import RequestSourcediffFilesFile
from .request_sourcediff_issue import RequestSourcediffIssue
from .request_sourcediff_new import RequestSourcediffNew
from .request_sourcediff_old import RequestSourcediffOld
class RequestSourcediff(XmlModel):
XML_TAG = "sourcediff"
key: str = Field(
xml_attribute=True,
)
old: Optional[RequestSourcediffOld] = Field(
)
new: Optional[RequestSourcediffNew] = Field(
)
files_list: List[RequestSourcediffFilesFile] = Field(
xml_name="files",
xml_wrapped=True,
)
issue_list: Optional[List[RequestSourcediffIssue]] = Field(
xml_name="issues",
xml_wrapped=True,
xml_item_name="issue",
)

View File

@ -0,0 +1,13 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
class RequestSourcediffFileDiff(XmlModel):
XML_TAG = "diff"
lines: int = Field(
xml_attribute=True,
)
text: str = Field(
xml_set_text=True,
)

View File

@ -0,0 +1,17 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
class RequestSourcediffFileNew(XmlModel):
XML_TAG = "new"
name: str = Field(
xml_attribute=True,
)
md5: str = Field(
xml_attribute=True,
)
size: int = Field(
xml_attribute=True,
)

View File

@ -0,0 +1,17 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
class RequestSourcediffFileOld(XmlModel):
XML_TAG = "old"
name: str = Field(
xml_attribute=True,
)
md5: str = Field(
xml_attribute=True,
)
size: int = Field(
xml_attribute=True,
)

View File

@ -0,0 +1,21 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
from .request_sourcediff_file_diff import RequestSourcediffFileDiff
from .request_sourcediff_file_new import RequestSourcediffFileNew
from .request_sourcediff_file_old import RequestSourcediffFileOld
class RequestSourcediffFilesFile(XmlModel):
XML_TAG = "file"
state: str = Field(
xml_attribute=True,
)
old: Optional[RequestSourcediffFileOld] = Field(
)
new: Optional[RequestSourcediffFileNew] = Field(
)
diff: RequestSourcediffFileDiff = Field(
)

View File

@ -0,0 +1,25 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
class RequestSourcediffIssue(XmlModel):
XML_TAG = "issue"
state: str = Field(
xml_attribute=True,
)
tracker: str = Field(
xml_attribute=True,
)
name: str = Field(
xml_attribute=True,
)
label: str = Field(
xml_attribute=True,
)
url: str = Field(
xml_attribute=True,
)

View File

@ -0,0 +1,21 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
class RequestSourcediffNew(XmlModel):
XML_TAG = "new"
project: str = Field(
xml_attribute=True,
)
package: str = Field(
xml_attribute=True,
)
rev: str = Field(
xml_attribute=True,
)
srcmd5: str = Field(
xml_attribute=True,
)

View File

@ -0,0 +1,21 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
class RequestSourcediffOld(XmlModel):
XML_TAG = "old"
project: str = Field(
xml_attribute=True,
)
package: str = Field(
xml_attribute=True,
)
rev: str = Field(
xml_attribute=True,
)
srcmd5: str = Field(
xml_attribute=True,
)

View File

@ -0,0 +1,33 @@
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
from .enums import RequestStates
class RequestState(XmlModel):
XML_TAG = "state"
name: RequestStates = Field(
xml_attribute=True,
)
who: Optional[str] = Field(
xml_attribute=True,
)
when: Optional[str] = Field(
xml_attribute=True,
)
created: Optional[str] = Field(
xml_attribute=True,
)
superseded_by: Optional[int] = Field(
xml_attribute=True,
)
approver: Optional[str] = Field(
xml_attribute=True,
)
comment: Optional[str] = Field(
)