diff --git a/doc/source/api.rst b/doc/source/api.rst index 4a5a726..7beb3ce 100644 --- a/doc/source/api.rst +++ b/doc/source/api.rst @@ -43,6 +43,20 @@ Extensions .. automodule:: osctiny.extensions.search :members: +.. automodule:: osctiny.extensions.staging + :members: + +Models +------ + +.. automodule:: osctiny.models + :members: + :undoc-members: + +.. automodule:: osctiny.models.staging + :members: + :undoc-members: + Utilities --------- diff --git a/doc/source/conf.py b/doc/source/conf.py index 1c23358..7e25f4c 100644 --- a/doc/source/conf.py +++ b/doc/source/conf.py @@ -15,18 +15,19 @@ import os import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath('.')))) +from osctiny import __version__ # -- Project information ----------------------------------------------------- project = 'OSC Tiny' -copyright = '2022, Andreas Hasenkopf' -author = 'Andreas Hasenkopf' +copyright = '2024, SUSE' +author = 'SUSE MAE Team' # The short X.Y version -version = '0.6.2' +version = __version__ # The full version, including alpha/beta/rc tags -release = '0.6.2' +release = __version__ # -- General configuration --------------------------------------------------- diff --git a/doc/source/quickstart.rst b/doc/source/quickstart.rst index 1c7291e..73dc6ad 100644 --- a/doc/source/quickstart.rst +++ b/doc/source/quickstart.rst @@ -15,7 +15,7 @@ from GitHub using ``pip``: .. code-block:: bash - pip install git+https://github.com/crazyscientist/osc-tiny.git + pip install git+https://github.com/SUSE/osc-tiny.git Strong authentication ^^^^^^^^^^^^^^^^^^^^^ diff --git a/osctiny/extensions/buildresults.py b/osctiny/extensions/buildresults.py index 1b80d51..b818bfd 100644 --- a/osctiny/extensions/buildresults.py +++ b/osctiny/extensions/buildresults.py @@ -111,6 +111,28 @@ def get_package_list(self, project, repo, arch): return self.osc.get_objectified_xml(response) + def get_status_and_build_id(self, project, repo, arch): + """ + Get build status and build ID + + :param project: Project name + :param repo: Repository name + :param arch: Architecture name + :return: Objectified XML element + :rtype: lxml.objectify.ObjectifiedElement + + .. versionadded:: {{ NEXT_RELEASE }} + """ + response = self.osc.request( + method="GET", + url=urljoin(self.osc.url, "{}/{}/{}/{}".format( + self.base_path, project, repo, arch + )), + params={"view": "status"} + ) + + return self.osc.get_objectified_xml(response) + def get_binary_list(self, project, repo, arch, package, **params): """ Get a list of built RPMs diff --git a/osctiny/extensions/projects.py b/osctiny/extensions/projects.py index c52c756..e9eef49 100644 --- a/osctiny/extensions/projects.py +++ b/osctiny/extensions/projects.py @@ -45,7 +45,7 @@ def get_list(self, deleted=False): def get_meta(self, project, rev=None): """ Get project metadata - + .. versionchanged:: 0.8.0 Added the ``rev`` parameter diff --git a/osctiny/extensions/staging.py b/osctiny/extensions/staging.py new file mode 100644 index 0000000..45e606c --- /dev/null +++ b/osctiny/extensions/staging.py @@ -0,0 +1,411 @@ +""" +Staging extension +----------------- + +This extension provides access to the staging workflow of OpenBuildService. + +.. seealso:: + + https://openbuildservice.org/help/manuals/obs-user-guide/cha.obs.stagingworkflow + + https://openbuildservice.org/help/manuals/obs-user-guide/cha.obs.best-practices.webuiusage#staging_how_to + +.. versionadded:: {{ NEXT_RELEASE }} +""" +import typing +from urllib.parse import urljoin + +from lxml.objectify import ObjectifiedElement, Element, SubElement + +from ..models.staging import ExcludedRequest, CheckReport +from ..utils.base import ExtensionBase + + +class Staging(ExtensionBase): + """ + Osc extension for interacting with staging workflows + """ + base_path_staging = "/staging" + base_path_status = "/status_reports" + + def get_backlog(self, project: str) -> ObjectifiedElement: + """ + List the requests in the staging backlog + + :param project: Project name + :return: Objectified XML element + :rtype: lxml.objectify.ObjectifiedElement + """ + response = self.osc.request( + method="GET", + url=urljoin(self.osc.url, "{}/{}/backlog".format(self.base_path_staging, project)) + ) + + return self.osc.get_objectified_xml(response) + + def get_excluded_requests(self, project: str) -> ObjectifiedElement: + """ + List the requests excluded from a staging workflow + + :param project: Project name + :return: Objectified XML element + :rtype: lxml.objectify.ObjectifiedElement + """ + response = self.osc.request( + method="GET", + url=urljoin(self.osc.url, "{}/{}/excluded_requests".format(self.base_path_staging, + project)) + ) + + return self.osc.get_objectified_xml(response) + + def set_excluded_requests(self, project: str, *requests: ExcludedRequest) -> bool: + """ + Exclude requests from the staging workflow. + + :param project: Project name + :param requests: Requests to exclude with optional reason/description + :return: ``True``, if successful. + :raises HTTPError: if comment was not saved correctly. The raised exception contains the + full response object and API response. + """ + excluded_requests = Element("excluded_requests") + for request in requests: + SubElement(excluded_requests, "request", **request.asdict()) + + response = self.osc.request( + method="POST", + url=urljoin(self.osc.url, "{}/{}/excluded_requests".format(self.base_path_staging, + project)), + data=excluded_requests + ) + parsed = self.osc.get_objectified_xml(response) + if response.status_code == 200 and parsed.get("code") == "ok": + return True + + return False + + def delete_excluded_requests(self, project: str, *requests: ExcludedRequest) -> bool: + """ + Remove requests from list of excluded requests + + :param project: Project name + :param requests: Requests to exclude with optional reason/description + :return: ``True``, if successful. + :raises HTTPError: if comment was not saved correctly. The raised exception contains the + full response object and API response. + """ + excluded_requests = Element("excluded_requests") + for request in requests: + SubElement(excluded_requests, "request", **request.asdict()) + + response = self.osc.request( + method="DELETE", + url=urljoin(self.osc.url, "{}/{}/excluded_requests".format(self.base_path_staging, + project)), + data=excluded_requests + ) + parsed = self.osc.get_objectified_xml(response) + if response.status_code == 200 and parsed.get("code") == "ok": + return True + + return False + + def get_staging_projects(self, project: str) -> ObjectifiedElement: + """ + List all the staging projects of a staging workflow. + + :param project: Project name + :return: Objectified XML element + :rtype: lxml.objectify.ObjectifiedElement + """ + response = self.osc.request( + method="GET", + url=urljoin(self.osc.url, + "{}/{}/staging_projects".format(self.base_path_staging, project)) + ) + + return self.osc.get_objectified_xml(response) + + # pylint: disable=too-many-arguments + def get_status(self, project: str, staging_project: str, requests: bool = False, + status: bool = False, history: bool = False) -> ObjectifiedElement: + """ + Get the overall state of a staging project + + :param project: Project name + :param staging_project: Staging project name + :param requests: Include statistics about staged, untracked and obsolete requests as well as + missing reviews + :param status: Include the overall state + :param history: Include the history of the staging project + :return: Objectified XML element + :rtype: lxml.objectify.ObjectifiedElement + """ + response = self.osc.request( + method="GET", + url=urljoin(self.osc.url, + "{}/{}/staging_projects/{}".format(self.base_path_staging, project, + staging_project)), + params={"requests": requests, "status": status, "history": history} + ) + + return self.osc.get_objectified_xml(response) + + def accept(self, project: str, staging_project: str) -> bool: + """ + This accepts all staged requests and sets the project state back to 'empty' + + :param project: Project name + :param staging_project: Staging project name + :return: ``True``, if successful. + :raises HTTPError: if comment was not saved correctly. The raised exception contains the + full response object and API response. + """ + response = self.osc.request( + method="POST", + url=urljoin(self.osc.url, "{}/{}/staging_projects/{}/accept".format( + self.base_path_staging, project, staging_project)) + ) + + parsed = self.osc.get_objectified_xml(response) + if response.status_code == 200 and parsed.get("code") == "ok": + return True + + return False + + def get_staged_requests(self, project: str, staging_project: str) ->ObjectifiedElement: + """ + List all the staged requests of a staging project + + :param project: Project name + :param staging_project: Staging project name + :return: Objectified XML element + :rtype: lxml.objectify.ObjectifiedElement + """ + response = self.osc.request( + method="GET", + url=urljoin(self.osc.url, + "{}/{}/staging_projects/{}/staged_requests".format( + self.base_path_staging, project, staging_project)), + ) + + return self.osc.get_objectified_xml(response) + + def add_staged_requests(self, project: str, staging_project: str, *request_ids: int) -> bool: + """ + Add requests to the staging project. + + :param project: Project name + :param staging_project: Staging project name + :param request_ids: Request IDs + :return: ``True``, if successful. + :raises HTTPError: if comment was not saved correctly. The raised exception contains the + full response object and API response. + """ + requests = Element("requests") + for request_id in request_ids: + SubElement(requests, "request", id=str(request_id)) + response = self.osc.request( + method="POST", + url=urljoin(self.osc.url, "{}/{}/staging_projects/{}/staged_requests".format( + self.base_path_staging, project, staging_project)), + data=requests + ) + parsed = self.osc.get_objectified_xml(response) + if response.status_code == 200 and parsed.get("code") == "ok": + return True + + return False + + def delete_staged_requests(self, project: str, staging_project: str, *request_ids: int) -> bool: + """ + Delete requests from the staging project + + :param project: Project name + :param staging_project: Staging project name + :param request_ids: Request IDs + :return: ``True``, if successful. + :raises HTTPError: if comment was not saved correctly. The raised exception contains the + full response object and API response. + """ + requests = Element("requests") + for request_id in request_ids: + SubElement(requests, "request", id=str(request_id)) + response = self.osc.request( + method="DELETE", + url=urljoin(self.osc.url, "{}/{}/staging_projects/{}/staged_requests".format( + self.base_path_staging, project, staging_project)), + data=requests + ) + parsed = self.osc.get_objectified_xml(response) + if response.status_code == 200 and parsed.get("code") == "ok": + return True + + return False + + def get_required_checks(self, project: str, repo: typing.Optional[str] = None, + arch: typing.Optional[str] = None) -> ObjectifiedElement: + """ + Get list of required checks + + If `repo`` and ``arch`` are specified, required checks from the built repository are + returned. If only ``repo`` is specified, required checks from the repository are + returned. Otherwise, required checks from the project are returned + + :param project: (Staging) project name + :param repo: Repository name (optional) + :param arch: Architecture name (optional) + :return: Objectified XML element + :rtype: lxml.objectify.ObjectifiedElement + """ + if repo and arch: + url_path = "{}/built_repositories/{}/{}/{}/required_checks".format( + self.base_path_status, project, repo, arch + ) + elif repo: + url_path = "{}/repositories/{}/{}/required_checks".format( + self.base_path_status, project, repo + ) + else: + url_path = "{}/projects/{}/required_checks".format(self.base_path_status, project) + + response = self.osc.request( + method="GET", + url=urljoin(self.osc.url, url_path), + ) + + return self.osc.get_objectified_xml(response) + + def set_required_checks(self, project: str, checks: typing.List[str], + repo: typing.Optional[str] = None, + arch: typing.Optional[str] = None) -> bool: + """ + Submit a new or modified required checks list + + If ``repo`` and ``arch`` are specified, required checks of built repository are + updated. If only ``repo`` is specified, required checks of the repository are updated. + Otherwise, required checks of the project are updated. + + :param project: (Staging) project name + :param checks: List of check names + :param repo: Repository name (optional) + :param arch: Architecture name (optional) + :return: ``True``, if successful. + :raises HTTPError: if comment was not saved correctly. The raised exception contains the + full response object and API response. + """ + # pylint: disable=protected-access + kwargs = {"project": project} + if repo and arch: + url_path = "{}/built_repositories/{}/{}/{}/required_checks".format( + self.base_path_status, project, repo, arch + ) + kwargs.update({"repository": repo, "architecture": arch}) + elif repo: + url_path = "{}/repositories/{}/{}/required_checks".format( + self.base_path_status, project, repo + ) + kwargs["repository"] = repo + else: + url_path = "{}/projects/{}/required_checks".format(self.base_path_status, project) + + required_checks = Element("required_checks") + for check in checks: + elem = SubElement(required_checks, "name") + elem._setText(check) + + response = self.osc.request( + method="POST", + url=urljoin(self.osc.url, url_path), + data=required_checks + ) + parsed = self.osc.get_objectified_xml(response) + if response.status_code == 200 and parsed.get("code") == "ok": + return True + + return False + + def get_status_report(self, project: str, repo: str, build_id: str, + arch: typing.Optional[str] = None) -> ObjectifiedElement: + """ + Get list of checks + + If ``arch`` is specified, status report for built project is retrieved. Otherwise, status + report for published project is retrieved. + + :param project: (Staging) project name + :param repo: Repository name + :param build_id: Build ID (Can be obtained via + :py:meth:`osctiny.extensions.buildresults.Build.get_status_and_build_id`) + :param arch: Architecture name + :return: Objectified XML element + :rtype: lxml.objectify.ObjectifiedElement + """ + if arch: + url_path = "{}/built/{}/{}/{}/reports/{}".format( + self.base_path_status, project, repo, arch, build_id + ) + else: + url_path = "{}/published/{}/{}/reports/{}".format( + self.base_path_status, project, repo, build_id + ) + + response = self.osc.request( + method="GET", + url=urljoin(self.osc.url, url_path) + ) + + return self.osc.get_objectified_xml(response) + + def set_status_report(self, project: str, repo: str, build_id: str, report: CheckReport, + arch: typing.Optional[str] = None) -> bool: + """ + Submit a check to a status report + + If ``arch`` is specified, the status report for built project is set. Otherwise, the status + report for published project is set. + + :param project: (Staging) project name + :param repo: Repository name + :param build_id: Build ID (Can be obtained via + :py:meth:`osctiny.extensions.buildresults.Build.get_status_and_build_id`) + :param report: The status upate + :param arch: Architecture name + :return: ``True``, if successful. + :raises HTTPError: if comment was not saved correctly. The raised exception contains the + full response object and API response. + """ + # pylint: disable=protected-access + check = Element("check", + name=report.name, + required="true" if report.required else "false") + state = SubElement(check, "state") + state._setText(report.state.value) + if report.short_description: + short_desc = SubElement(check, "short_description") + short_desc._setText(report.short_description) + if report.url: + url = SubElement(check, "url") + url._setText(report.url) + + if arch: + url_path = "{}/built/{}/{}/{}/reports/{}".format( + self.base_path_status, project, repo, arch, build_id + ) + else: + url_path = "{}/published/{}/{}/reports/{}".format( + self.base_path_status, project, repo, build_id + ) + + response = self.osc.request( + method="POST", + url=urljoin(self.osc.url, url_path), + data=check + ) + + parsed = self.osc.get_objectified_xml(response) + if response.status_code == 200 and parsed.get("code") == "ok": + return True + + return False diff --git a/osctiny/models/__init__.py b/osctiny/models/__init__.py new file mode 100644 index 0000000..23f26a3 --- /dev/null +++ b/osctiny/models/__init__.py @@ -0,0 +1,11 @@ +""" +Common model/type definitions +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +""" +from io import BufferedReader, BytesIO, StringIO +import typing + +from lxml.objectify import ObjectifiedElement + + +ParamsType = typing.Union[bytes, str, StringIO, BytesIO, BufferedReader, dict, ObjectifiedElement] diff --git a/osctiny/models/staging.py b/osctiny/models/staging.py new file mode 100644 index 0000000..7169cf8 --- /dev/null +++ b/osctiny/models/staging.py @@ -0,0 +1,34 @@ +""" +Models for Staging +^^^^^^^^^^^^^^^^^^ +""" +# pylint: disable=missing-class-docstring,missing-function-docstring +import enum +import typing + + +class ExcludedRequest(typing.NamedTuple): + id: int + description: typing.Optional[str] = None + + def asdict(self) -> typing.Dict[str, str]: + d = {"id": str(self.id)} + if self.description: + d["description"] = self.description + + return d + + +class CheckState(enum.Enum): + PENDING = "pending" + ERROR = "error" + FAILURE = "failure" + SUCCESS = "success" + + +class CheckReport(typing.NamedTuple): + name: str + required: bool + state: CheckState + short_description: typing.Optional[str] = None + url: typing.Optional[str] = None diff --git a/osctiny/osc.py b/osctiny/osc.py index e1d7b26..94779fc 100644 --- a/osctiny/osc.py +++ b/osctiny/osc.py @@ -20,6 +20,8 @@ from urllib.parse import quote, parse_qs, urlparse import warnings +from lxml.etree import tostring +from lxml.objectify import ObjectifiedElement from requests import Session, Request from requests.auth import HTTPBasicAuth from requests.cookies import RequestsCookieJar, cookiejar_from_dict @@ -35,7 +37,9 @@ from .extensions.projects import Project from .extensions.bs_requests import Request as BsRequest from .extensions.search import Search +from .extensions.staging import Staging from .extensions.users import Group, Person +from .models import ParamsType from .utils.auth import HttpSignatureAuth from .utils.backports import cached_property from .utils.conf import BOOLEAN_PARAMS, get_credentials, get_cookie_jar @@ -87,6 +91,8 @@ class Osc: - :py:attr:`origins` * - :py:class:`osctiny.extensions.attributes.Attribute` - :py:attr:`attributes` + * - :py:class:`osctiny.extensions.staging.Staging` + - :py:attr:`staging` :param url: API URL of a BuildService instance :param username: Username @@ -120,7 +126,10 @@ class Osc: .. versionchanged:: 0.8.0 * Removed the ``cache`` parameter - * Added the ``attributes`` extensions + * Added the ``attributes`` extension + + .. versionchanged:: {{NEXT_RELEASE}} + * Added the ``staging`` extension .. _SSL Cert Verification: http://docs.python-requests.org/en/master/user/advanced/ @@ -163,6 +172,7 @@ def __init__(self, url: typing.Optional[str] = None, username: typing.Optional[s self.projects = Project(osc_obj=self) self.requests = BsRequest(osc_obj=self) self.search = Search(osc_obj=self) + self.staging = Staging(osc_obj=self) self.users = Person(osc_obj=self) def __del__(self): @@ -222,7 +232,7 @@ def parser(self): Explicit parser instance .. versionchanged:: 0.8.0 - Content moved to :py:fun:`osctiny.utils.xml.get_xml_parser` + Content moved to :py:func:`osctiny.utils.xml.get_xml_parser` """ return get_xml_parser() @@ -351,9 +361,8 @@ def get_boolean_params(self, url: str, method: str) -> typing.Tuple[str]: return () - def handle_params(self, url: str, method: str, - params: typing.Union[bytes, str, StringIO, BytesIO, BufferedReader, dict]) \ - -> bytes: + def handle_params(self, url: str, method: str, params: ParamsType) \ + -> bytes: # pylint: disable=too-many-return-statements """ Translate request parameters to API conform format @@ -369,7 +378,6 @@ def handle_params(self, url: str, method: str, /9715 :param params: Request parameter - :type params: dict or str or io.BufferedReader :param url: URL to which the parameters will be sent :type url: str :param method: HTTP method to send request @@ -380,6 +388,10 @@ def handle_params(self, url: str, method: str, .. versionchanged:: 0.7.3 Added the ``url`` and ``method`` parameters + + .. versionchanged:: {{ NEXT_RELEASE }} + + Instances of ``ObjectifiedElement`` are accepted for argument ``params`` """ if isinstance(params, bytes): return params @@ -395,6 +407,9 @@ def handle_params(self, url: str, method: str, params.seek(0) return params + if isinstance(params, ObjectifiedElement): + return tostring(params, encoding="utf-8", xml_declaration=True) + if not isinstance(params, dict): return {} @@ -471,6 +486,6 @@ def get_objectified_xml(self, response): .. versionchanged:: 0.8.0 - Content moved to :py:fun:`osctiny.utils.xml.get_objectified_xml` + Content moved to :py:func:`osctiny.utils.xml.get_objectified_xml` """ return get_objectified_xml(response=response) diff --git a/osctiny/tests/test_staging.py b/osctiny/tests/test_staging.py new file mode 100644 index 0000000..415ffe7 --- /dev/null +++ b/osctiny/tests/test_staging.py @@ -0,0 +1,248 @@ +import re + +import responses + +from osctiny.models.staging import ExcludedRequest, CheckState, CheckReport + +from .base import OscTest + + +class StagingTest(OscTest): + @staticmethod + def _mock_generic_request(): + responses.add(method=responses.GET, + url=re.compile("http://api.example.com/(staging|status_reports)/.*"), + body="Hello World", + status=200) + + @responses.activate + def test_get_backlog(self): + self._mock_generic_request() + response = self.osc.staging.get_backlog("Dummy:Project") + self.assertEqual(response.tag, "dummy") + + @responses.activate + def test_get_excluded_requests(self): + self._mock_generic_request() + response = self.osc.staging.get_excluded_requests("Dummy:Project") + self.assertEqual(response.tag, "dummy") + + @responses.activate + def test_set_excluded_requests(self): + def callback(request): + element = self.osc.get_objectified_xml(response=request.body) + self.assertEqual(element.request[0].get("id"), '1') + self.assertEqual(element.request[0].get("description"), "Foo Bar") + self.assertEqual(element.request[1].get("id"), '2') + self.assertEqual(element.request[1].get("description"), "Hello World") + self.assertEqual(element.request[2].get("id"), '3') + return 200, {}, "" + + self.mock_request( + method="POST", + url="http://api.example.com/staging/Dummy:Project/excluded_requests", + callback=callback + ) + + result = self.osc.staging.set_excluded_requests("Dummy:Project", + ExcludedRequest(1, "Foo Bar"), + ExcludedRequest(2, "Hello World"), + ExcludedRequest(3)) + self.assertTrue(result) + + @responses.activate + def test_delete_excluded_requests(self): + def callback(request): + element = self.osc.get_objectified_xml(response=request.body) + self.assertEqual(element.request[0].get("id"), '1') + return 200, {}, "" + + self.mock_request( + method="DELETE", + url="http://api.example.com/staging/Dummy:Project/excluded_requests", + callback=callback + ) + + result = self.osc.staging.delete_excluded_requests("Dummy:Project", ExcludedRequest(1)) + self.assertTrue(result) + + @responses.activate + def test_get_staging_projects(self): + self._mock_generic_request() + response = self.osc.staging.get_staging_projects("Dummy:Project") + self.assertEqual(response.tag, "dummy") + + @responses.activate + def test_get_status(self): + self._mock_generic_request() + response = self.osc.staging.get_status("Dummy:Project", "Dummy:Project:Staging:A") + self.assertEqual(response.tag, "dummy") + + @responses.activate + def test_accept(self): + responses.add(method="POST", + url="http://api.example.com/staging/Dummy:Project/staging_projects/Dummy:Project:Staging:A/accept", + status=200, + body="") + self.assertTrue(self.osc.staging.accept("Dummy:Project", "Dummy:Project:Staging:A")) + + @responses.activate + def test_get_staged_requests(self): + self._mock_generic_request() + response = self.osc.staging.get_staged_requests("Dummy:Project", "Dummy:Project:Staging:A") + self.assertEqual(response.tag, "dummy") + + @responses.activate + def test_add_staged_requests(self): + def callback(request): + element = self.osc.get_objectified_xml(request.body) + self.assertEqual(3, len(element.request)) + self.assertEqual(["1", "2", "3"], [elem.get("id") for elem in element.request]) + return 200, {}, "" + + responses.add_callback(method="POST", + url="http://api.example.com/staging/Dummy:Project/staging_projects/Dummy:Project:Staging:A/staged_requests", + callback=callback) + result = self.osc.staging.add_staged_requests("Dummy:Project", "Dummy:Project:Staging:A", + 1, 2, 3) + self.assertTrue(result) + + @responses.activate + def test_delete_staged_requests(self): + def callback(request): + element = self.osc.get_objectified_xml(request.body) + self.assertEqual(3, len(element.request)) + self.assertEqual(["1", "2", "3"], [elem.get("id") for elem in element.request]) + return 200, {}, "" + + responses.add_callback(method="DELETE", + url="http://api.example.com/staging/Dummy:Project/staging_projects/Dummy:Project:Staging:A/staged_requests", + callback=callback) + result = self.osc.staging.delete_staged_requests("Dummy:Project", "Dummy:Project:Staging:A", + 1, 2, 3) + self.assertTrue(result) + + @responses.activate + def test_get_required_checks(self): + responses.add( + method="GET", + url="http://api.example.com/status_reports/built_repositories/Dummy:Project/repo/x86_64/required_checks", + body="", + status=200 + ) + responses.add( + method="GET", + url="http://api.example.com/status_reports/repositories/Dummy:Project/repo/required_checks", + body="", + status=200 + ) + responses.add( + method="GET", + url="http://api.example.com/status_reports/projects/Dummy:Project/required_checks", + body="", + status=200 + ) + + with self.subTest("Repo + Arch"): + response = self.osc.staging.get_required_checks("Dummy:Project", "repo", "x86_64") + self.assertEqual(response.tag, "dummyrepoarch") + + with self.subTest("Repo"): + response = self.osc.staging.get_required_checks("Dummy:Project", "repo") + self.assertEqual(response.tag, "dummyrepo") + + with self.subTest("Project"): + response = self.osc.staging.get_required_checks("Dummy:Project") + self.assertEqual(response.tag, "dummyproject") + + @responses.activate + def test_set_required_checks(self): + def callback(request): + elem = self.osc.get_objectified_xml(request.body) + + if "/projects/" in request.url: + self.assertTrue(all("project" in child.text for child in elem.name)) + if "/repositories/" in request.url: + self.assertTrue(all("repo" in child.text for child in elem.name)) + if "/built_repositories/" in request.url: + self.assertTrue(all("built" in child.text for child in elem.name)) + + return 200, {}, "" + + responses.add_callback( + method="POST", + url=re.compile("http://api.example.com/status_reports/(projects|repositories|built_repositories)/.*"), + callback=callback + ) + + with self.subTest("Repo + Arch"): + result = self.osc.staging.set_required_checks( + "Dummy:Project", [f"built-{i}" for i in range(1, 3)], "repo", "x86_64" + ) + self.assertTrue(result) + + with self.subTest("Repo"): + result = self.osc.staging.set_required_checks( + "Dummy:Project", [f"repo-{i}" for i in range(1, 3)], "repo" + ) + self.assertTrue(result) + + with self.subTest("Project"): + result = self.osc.staging.set_required_checks( + "Dummy:Project", [f"project-{i}" for i in range(1, 3)] + ) + self.assertTrue(result) + + @responses.activate + def test_get_status_report(self): + responses.add( + method="GET", + url="http://api.example.com/status_reports/built/Dummy:Project/repo/x86_64/reports/1", + body="", + status=200 + ) + responses.add( + method="GET", + url="http://api.example.com/status_reports/published/Dummy:Project/repo/reports/1", + body="", + status=200 + ) + + with self.subTest("Repo + Arch"): + response = self.osc.staging.get_status_report("Dummy:Project", "repo", "1", "x86_64") + self.assertEqual(response.tag, "dummyarch") + + with self.subTest("Repo"): + response = self.osc.staging.get_status_report("Dummy:Project", "repo", "1") + self.assertEqual(response.tag, "dummyrepo") + + @responses.activate + def test_set_status_report(self): + report = CheckReport( + name="dummy-check", + required=False, + state=CheckState.FAILURE, + short_description="Lorem ipsum dolor sit", + url="http://example.com/lorem-ipsum" + ) + + def callback(request): + elem = self.osc.get_objectified_xml(request.body) + self.assertEqual(report.name, elem.get("name")) + self.assertEqual("false", elem.get("required")) + self.assertEqual(elem.state.text, report.state.value) + self.assertEqual(elem.short_description.text, report.short_description) + self.assertEqual(elem.url.text, report.url) + + return 200, {}, "" + + responses.add_callback( + method="POST", + url=re.compile("http://api.example.com/status_reports/(published|built)"), + callback=callback + ) + + with self.subTest("Built"): + result = self.osc.staging.set_status_report("Dummy:Project", "repo", "id-1", report, + "x86_64") + self.assertTrue(result) diff --git a/osctiny/utils/xml.py b/osctiny/utils/xml.py index af27e2e..8b0b42e 100644 --- a/osctiny/utils/xml.py +++ b/osctiny/utils/xml.py @@ -30,7 +30,7 @@ def get_xml_parser() -> XMLParser: return THREAD_LOCAL.parser -def get_objectified_xml(response: typing.Union[Response, str]) -> ObjectifiedElement: +def get_objectified_xml(response: typing.Union[Response, str, bytes]) -> ObjectifiedElement: """ Return API response as an XML object @@ -46,11 +46,15 @@ def get_objectified_xml(response: typing.Union[Response, str]) -> ObjectifiedEle Carved out from ``Osc`` class + .. versionchanged:: {{ NEXT_RELEASE }} + + Accepts also bytes + :param response: An API response or XML string :rtype response: :py:class:`requests.Response` :return: :py:class:`lxml.objectify.ObjectifiedElement` """ - if isinstance(response, str): + if isinstance(response, (str, bytes)): text = response elif isinstance(response, Response): text = response.text diff --git a/pylint.rc b/pylint.rc index fbeafae..ba71aa9 100644 --- a/pylint.rc +++ b/pylint.rc @@ -205,8 +205,7 @@ min-public-methods=2 [EXCEPTIONS] # Exceptions that will emit a warning when caught. -overgeneral-exceptions=BaseException, - Exception +overgeneral-exceptions= [FORMAT]