diff --git a/sw360/__init__.py b/sw360/__init__.py index ecd246e..0989cfd 100644 --- a/sw360/__init__.py +++ b/sw360/__init__.py @@ -9,6 +9,8 @@ __version__ = (1, 11, 2) +from .sorting import (BaseSortMixin, ProjectSortColumn, ReleaseSortColumn, + SortParam) from .sw360_api import SW360 from .sw360error import SW360Error from .sw360keycloak import SW360Keycloak @@ -18,5 +20,9 @@ "SW360", "SW360Error", "SW360OAuth2", - "SW360Keycloak" + "SW360Keycloak", + "ProjectSortColumn", + "ReleaseSortColumn", + "SortParam", + "BaseSortMixin", ] diff --git a/sw360/base.py b/sw360/base.py index 4856b3b..b2d965d 100644 --- a/sw360/base.py +++ b/sw360/base.py @@ -7,13 +7,21 @@ # SPDX-License-Identifier: MIT # ------------------------------------------------------------------------------- +import json +import logging +import os +from http import HTTPStatus from typing import Any, Dict, List, Optional, Tuple, Union from urllib.parse import urlencode import requests +from .sorting import SortParam from .sw360error import SW360Error +logger = logging.getLogger(__name__) +logger.setLevel(logging.WARNING) + class BaseMixin(): """Python interface to the Siemens SW360 platform @@ -25,16 +33,19 @@ class BaseMixin(): token_type is "Bearer" for an OAuth workflow and "Token" for tokens generated via the SW360 UI. - :param url: URL of the SW360 instance - :param token: The SW360 REST API token (the cryptic string without + :ivar url: URL of the SW360 instance + :ivar token: The SW360 REST API token (the cryptic string without "Authorization:" and `token_type`). - :param oauth2: flag indicating whether this is an OAuth2 token + :ivar oauth2: flag indicating whether this is an OAuth2 token + :ivar default_batch_size: Default size of batch to use while fetching all items from API :type url: string :type token: string :type oauth2: boolean + :type default_batch_size: int """ - def __init__(self, url: str, token: str, oauth2: bool = False) -> None: + def __init__(self, url: str, token: str, oauth2: bool = False, + default_batch_size: int = 50) -> None: """Constructor""" if url[-1] != "/": url += "/" @@ -47,6 +58,7 @@ def __init__(self, url: str, token: str, oauth2: bool = False) -> None: self.api_headers = {"Authorization": "Token " + token} self.force_no_session = False + self.default_batch_size = default_batch_size def api_get(self, url: str = "") -> Optional[Dict[str, Any]]: """Request `url` from REST API and return json answer. @@ -68,12 +80,76 @@ def api_get(self, url: str = "") -> Optional[Dict[str, Any]]: response = self.session.get(url) if response.ok: - if response.status_code == 204: # 204 = no content + if response.status_code == HTTPStatus.NO_CONTENT: return None return response.json() raise SW360Error(response, url) + def api_get_all(self, url: str, sort: Optional[SortParam] = None, + batch: int = -1, page: int = 0, + _data: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]: + """ + Retrieve all pages of data from the specified URL. + + :param url: The URL to request data from. + :param sort: The sort parameter to sort data by. + :param page: Page number to fetch + :param batch: How many rows to fetch at a time, use + `default_batch_size` if -1 + :param _data: Internal param for aggregating data. + :return: The combined JSON data from all pages. + :rtype: Optional[Dict[str, Any]] + """ + if _data is None: + _data = {} + if batch == -1: + batch = self.default_batch_size + + paginated_url = self._add_pagination(url, page, batch, sort) + resp = self.api_get(paginated_url) + if resp is not None and 'page' in resp: + total_pages = resp['page']['totalPages'] + # Clean up meta info + if '_links' in resp: + del resp['_links'] + del resp['page'] + # Update data and get next page + _data = self.__merge_responses(_data, resp) + if page + 1 < total_pages: + return self.api_get_all(url, sort, batch, page + 1, _data) + else: + # Clean up meta info + if resp is not None and '_links' in resp: + del resp['_links'] + _data = self.__merge_responses(_data, resp) + return _data + + def __merge_responses(self, previous: Dict[str, Any], + next: Optional[Dict[str, Any]]) -> Dict[str, Any]: + """ + Function to merge previous and next response of the same API + recursively. + :param previous: The previous response data. + :param next: The next response data. + :return: The merged response data. + """ + if next is None: + return previous + for key, value in next.items(): + if (key in previous and isinstance(previous[key], dict) and + isinstance(value, dict)): + previous[key] = self.__merge_responses(previous[key], value) + elif (key in previous and isinstance(previous[key], list) and + isinstance(value, list)): + previous[key].extend(value) + elif (key in previous and isinstance(previous[key], tuple) and + isinstance(value, tuple)): + previous[key].extend(value) + else: + previous[key] = value + return previous + def api_post_multipart(self, url: str = "", files: Dict[str, Any] = {}) -> Optional[requests.Response]: """ Send a multipart POST request to the specified URL with the provided file data. @@ -97,7 +173,7 @@ def api_post_multipart(self, url: str = "", files: Dict[str, Any] = {}) -> Optio response = self.session.post(url, files=files) if response.ok: - if response.status_code == 204: # 204 = no content + if response.status_code == HTTPStatus.NO_CONTENT: return None return response @@ -131,7 +207,7 @@ def api_post( response = self.session.post(url, json=json) if response.ok: - if response.status_code == 204: # 204 = no content + if response.status_code == HTTPStatus.NO_CONTENT: return None return response @@ -159,7 +235,7 @@ def api_patch(self, url: str = "", json: Any = {}) -> Optional[Dict[str, Any]]: response = self.session.patch(url, json=json) if response.ok: - if response.status_code == 204: # 204 = no content + if response.status_code == HTTPStatus.NO_CONTENT: return None if response.content: return response.json() @@ -187,7 +263,7 @@ def api_delete(self, url: str = "") -> Optional[requests.Response]: response = self.session.delete(url) if response.ok: - if response.status_code == 204: # 204 = no content + if response.status_code == HTTPStatus.NO_CONTENT: return None return response @@ -223,14 +299,37 @@ def _add_params(self, url: str, params: Dict[str, str]) -> str: query_string = urlencode(params) + if query_string == "": + return url + if "?" in url: return f"{url}&{query_string}" else: return f"{url}?{query_string}" + def _add_pagination(self, url: str, page: int, page_entries: int, + sort: Optional[SortParam] = None) -> str: + """ + Add pagination parameters to the GET request URL + :param url: URL to add params to + :param page: Page number to fetch + :param page_entries: Number of entries to fetch per request + :param sort: Sorting parameter (optional) + :return: URL with pagination parameters added + """ + + params = { + "page": str(page), + "page_entries": str(page_entries) + } + if sort is not None: + params["sort"] = str(sort) + + return self._add_params(url, params) + @classmethod def get_id_from_href(cls, href: str) -> str: - """"Extracts the identifier from the href and returns it + """Extracts the identifier from the href and returns it :param href: HAL href for a specific resource :type href: string (valid URL) @@ -241,3 +340,89 @@ def get_id_from_href(cls, href: str) -> str: pos = href.rfind("/") identifier = href[(pos + 1):] return identifier + + def _upload_resource_file(self, upload_file: str, + upload_type: str = "SOURCE", + upload_comment: str = "") -> Dict[str, str]: + """Upload `upload_file` as attachment to SW360 which can then be used + by various resources as attachment content. + `upload_type` can be: + "DOCUMENT" + "SOURCE" + "CLEARING_REPORT" + "COMPONENT_LICENSE_INFO_XML" + "SOURCE_SELF" + "BINARY" + "BINARY_SELF" + "LICENSE_AGREEMENT" + "README_OSS" + + API endpoint: POST /attachments + + :return: Returns the attachment content of the uploaded file which + can be used by other resources. + :raises SW360Error: if unable to extract content from the upload + """ + if not os.path.exists(upload_file): + raise SW360Error(message="ERROR: file not found: " + upload_file) + + filename = os.path.basename(upload_file) + url = self.url + "resource/api/attachments" + attachment_data = { + "filename": filename, + "createdComment": upload_comment, + "attachmentType": upload_type + } + + file_data = { + "files": (filename, open(upload_file, "rb"), "multipart/form-data"), + "attachment": ( + "", # dummy filename + json.dumps(attachment_data), + "application/json", + ), + } + response = self.api_post_multipart(url, files=file_data) + attachment_content = None + if response is not None: + if response.status_code == HTTPStatus.ACCEPTED: + logger.warning( + f"Attachment upload was accepted by {url} but might not be visible yet: {response.text}" + ) + if response.status_code == HTTPStatus.OK: + r = response.json() + if '_embedded' in r and 'sw360:attachments' in r['_embedded'] \ + and len(r['_embedded']['sw360:attachments']) == 1: + content = r['_embedded']['sw360:attachments'][0] + attachment_content = { + 'attachmentContentId': content['attachmentContentId'], + 'filename': content['filename'], + 'sha1': content['sha1'], + 'attachmentType': content['attachmentType'], + 'createdComment': content['createdComment'], + 'checkStatus': content['checkStatus'] + } + if not response.ok: + raise SW360Error(response, url) + if attachment_content is None: + raise SW360Error(response, url, + "Unable to fetch attachment content from the response.") + return attachment_content + + @staticmethod + def _get_attachments( + resource: Optional[Dict[str, Any]] + ) -> List[Dict[str, Any]]: + attachments: List[Dict[str, Any]] = [] + if resource is None: + return attachments + if 'attachments' in resource: + attachments.extend(resource['attachments']) + if '_embedded' in resource and \ + 'sw360:attachments' in resource['_embedded']: + attachments.extend(resource['_embedded']['sw360:attachments']) + # Remove meta properties like "_links" from attachments + for attachment in attachments: + if '_links' in attachment: + del attachment['_links'] + return attachments diff --git a/sw360/clearing.py b/sw360/clearing.py index c2892f9..f4d2549 100644 --- a/sw360/clearing.py +++ b/sw360/clearing.py @@ -39,8 +39,8 @@ def get_clearing_request_for_project(self, project_id: str) -> Optional[Dict[str API endpoint: GET /clearingrequest/project/{id} - :param request_id: the id of the clearing request to be requested - :type request_id: string + :param project_id: the id of the Project to fetch clearing requests for + :type project_id: string :return: a clearing request :rtype: JSON clearing request object :raises SW360Error: if there is a negative HTTP response diff --git a/sw360/components.py b/sw360/components.py index 6792d6e..794faa4 100644 --- a/sw360/components.py +++ b/sw360/components.py @@ -12,27 +12,70 @@ from typing import Any, Dict, List, Optional from .base import BaseMixin +from .sorting import ComponentSortColumn, SortParam from .sw360error import SW360Error class ComponentsMixin(BaseMixin): # return type List[Dict[str, Any]] | Optional[Dict[str, Any]] for Python 3.11 is good, # Union[List[Dict[str, Any]], Optional[Dict[str, Any]]] for lower Python versions is not good - def get_all_components(self, fields: str = "", page: int = -1, page_size: int = -1, - all_details: bool = False, - sort: str = "") -> Any: + + def __get_components_filtered( + self, url: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Any: + """ + Take a pre-generated URL of components endpoint, with filters applied. + Then call the API with appropriate pagination and sorting to get the + components. + + :param url: Components API URL with filters in the query + :type url: str + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the components (Sort by name if `None`) + :type sort: SortParam + :return: list of components + :rtype: list of JSON component objects + :raises SW360Error: if there is a negative HTTP response + """ + + full_url = self._add_params(url, {"luceneSearch": "true"}) + if page > -1 and page_size > -1: + full_url = self._add_pagination(full_url, page, page_size, sort) + + if page_size == -1: + resp = self.api_get_all(full_url, sort) + else: + resp = self.api_get(full_url) + + if (resp and + "_embedded" in resp and + "sw360:components" in resp["_embedded"]): + return resp["_embedded"]["sw360:components"] + + return [] + + def get_all_components( + self, fields: str = "", page: int = -1, page_size: int = -1, + all_details: bool = False, sort: Optional[SortParam] = None + ) -> Any: """Get information of about all components API endpoint: GET /components + :param fields: Comma-separated fields in the components object to fetch + :type fields: string :param page: page to retrieve :type page: int - :param page_size: page size to use + :param page_size: page size to use, `-1` to get all :type page_size: int - :param all_details: retrieve all component details (optional)) + :param all_details: retrieve all component details (optional) :type all_details: bool - :param sort: sort order for the components ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (Sort by name if `None`) + :type sort: SortParam :return: list of components :rtype: list of JSON component objects :raises SW360Error: if there is a negative HTTP response @@ -47,48 +90,32 @@ def get_all_components(self, fields: str = "", page: int = -1, page_size: int = if fields: params["fields"] = fields - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) + url_with_param = self._add_params(fullbase_url, params) - if sort: - params["sort"] = sort - - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) - if not resp: - return [] - - if "_embedded" not in resp: - return [] - - if "sw360:components" not in resp["_embedded"]: - return [] - - if page == -1: - return resp["_embedded"]["sw360:components"] + if sort is None: + sort = ComponentSortColumn.NAME.asc() - return resp + return self.__get_components_filtered(url_with_param, page, page_size, + sort) def get_components_by_type( - self, - component_type: str, - page: int = -1, - page_size: int = -1, - sort: str = "") -> List[Dict[str, Any]]: + self, component_type: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: """Get information of about all components for certain type - API endpoint: GET /components + API endpoint: GET /components?type= :param component_type: the type of the component to be requested, one - of INTERNAL, OSS, COTS, FREESOFTWARE, INNER_SOURCE, SERVICE + of INTERNAL, OSS, COTS, FREESOFTWARE, INNER_SOURCE, SERVICE, + CODE_SNIPPET, COTS_TRUSTED_SUPPLIER :type component_type: string :param page: page to retrieve :type page: int - :param page_size: page size to use + :param page_size: page size to use, `-1` to get all :type page_size: int - :param sort: sort order for the components ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (Sort by name if `None`) + :type sort: SortParam :return: list of components :rtype: list of JSON component objects :raises SW360Error: if there is a negative HTTP response @@ -99,20 +126,13 @@ def get_components_by_type( fullbase_url = self.url + "resource/api/components" params = {"type": component_type} - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) - - if sort: - params["sort"] = sort + url_with_param = self._add_params(fullbase_url, params) - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) + if sort is None: + sort = ComponentSortColumn.NAME.asc() - if resp and ("_embedded" in resp) and ("sw360:components" in resp["_embedded"]): - return resp["_embedded"]["sw360:components"] - - return [] + return self.__get_components_filtered(url_with_param, page, page_size, + sort) def get_component(self, component_id: str) -> Optional[Dict[str, Any]]: """Get information of about a component @@ -149,23 +169,21 @@ def get_component_by_url(self, component_url: str) -> Optional[Dict[str, Any]]: return resp def get_component_by_name( - self, - component_name: str, - page: int = -1, - page_size: int = -1, - sort: str = "") -> Optional[Dict[str, Any]]: + self, component_name: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Dict[str, Any]: """Get information of about a component - API endpoint: GET /components + API endpoint: GET /components?name= :param component_name: the name of the component to look for :type component_name: string :param page: page to retrieve :type page: int - :param page_size: page size to use + :param page_size: page size to use, `-1` to get all :type page_size: int - :param sort: sort order for the components ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (Sort by score if `None`) + :type sort: SortParam :return: list of components :rtype: list of JSON component objects :raises SW360Error: if there is a negative HTTP response @@ -176,16 +194,13 @@ def get_component_by_name( fullbase_url = self.url + "resource/api/components" params = {"name": component_name} - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) + url_with_param = self._add_params(fullbase_url, params) - if sort: - params["sort"] = sort + if sort is None: + sort = ComponentSortColumn.SCORE.asc() - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) - return resp + return self.__get_components_filtered(url_with_param, page, page_size, + sort) def get_components_by_external_id(self, ext_id_name: str, ext_id_value: str = "") -> List[Dict[str, Any]]: """Get components by external id. `ext_id_value` can be left blank to @@ -194,8 +209,8 @@ def get_components_by_external_id(self, ext_id_name: str, ext_id_value: str = "" API endpoint: GET /components :param ext_id_name: the name of the external id to look for - :param ext_id_value: the value of the external id to look for :type ext_id_name: string + :param ext_id_value: the value of the external id to look for :type ext_id_value: string :return: list of components :rtype: list of JSON component objects @@ -204,13 +219,12 @@ def get_components_by_external_id(self, ext_id_name: str, ext_id_value: str = "" if not ext_id_name: raise SW360Error(message="No external id name provided!") - resp = self.api_get( - self.url - + "resource/api/components/searchByExternalIds?" - + ext_id_name - + "=" - + ext_id_value - ) + fullbase_url = self.url + "resource/api/components/searchByExternalIds" + params = {ext_id_name: ext_id_value} + + url_with_param = self._add_params(fullbase_url, params) + + resp = self.api_get(url_with_param) if resp and ("_embedded" in resp) and ("sw360:components" in resp["_embedded"]): return resp["_embedded"]["sw360:components"] @@ -225,7 +239,8 @@ def create_new_component(self, name: str, description: str, component_type: str, :param name: name of the new component :param description: description of the new component :param component_type: type of the new component, one of - "INTERNAL", "OSS", "COTS", "FREESOFTWARE", "INNER_SOURCE", "SERVICE", "CODE_SNIPPET" + "INTERNAL", "OSS", "COTS", "FREESOFTWARE", "INNER_SOURCE", "SERVICE", + "CODE_SNIPPET", "COTS_TRUSTED_SUPPLIER" :param homepage: homepage url of the new component :param component_details: further component details as defined by SW360 REST API :type name: string @@ -369,3 +384,33 @@ def get_recent_components(self) -> Optional[List[Dict[str, Any]]]: return resp["_embedded"]["sw360:components"] return [] + + def upload_attachment_to_component( + self, component_id: str, upload_file: str, upload_type: str = "SOURCE", + upload_comment: str = "" + ) -> Optional[Dict[str, Any]]: + """Upload an attachment to a given Component. + + API endpoint: POST /attachments & PATCH /components/{id} + + :param component_id: the id of the Component + :type component_id: string + :param upload_file: path of the file to be uploaded + :type upload_file: string + :param upload_type: the type of the attachment + :type upload_type: string + :param upload_comment: a comment for the attachment + :type upload_comment: string + :raises SW360Error: if the component id is missing or there is a negative HTTP response + """ + if not component_id: + raise SW360Error(message="No component id provided!") + + attachment_content = self._upload_resource_file(upload_file, upload_type, upload_comment) + attachment_content['attachmentType'] = upload_type # Make sure the type is correct + attachment_content['createdComment'] = upload_comment # Override + + current_component = self.get_component(component_id) + attachments = self._get_attachments(current_component) + attachments.append(attachment_content) + return self.update_component({'attachments': attachments}, component_id) diff --git a/sw360/license.py b/sw360/license.py index f54143c..12cac92 100644 --- a/sw360/license.py +++ b/sw360/license.py @@ -14,6 +14,7 @@ import requests from .base import BaseMixin +from .sorting import LicenseSortColumn from .sw360error import SW360Error @@ -24,7 +25,6 @@ def create_new_license( fullName: str, text: str, checked: bool = False, - license_details: Dict[str, Any] = {}, ) -> Any: """Create a new license @@ -34,8 +34,8 @@ def create_new_license( :param fullName: descriptive license name :param text: license text :param checked: if license is checked - :type shortname: string - :type fullname: string + :type shortName: string + :type fullName: string :type text: string :type checked: bool :return: SW360 result @@ -53,10 +53,12 @@ def create_new_license( url = self.url + "resource/api/licenses" - license_details["shortName"] = shortName - license_details["fullName"] = fullName - license_details["text"] = text - license_details["checked"] = checked + license_details = { + "shortName": shortName, + "fullName": fullName, + "text": text, + "checked": checked, + } response = self.api_post(url, json=license_details) if response is not None: @@ -67,7 +69,7 @@ def create_new_license( def delete_license(self, license_shortname: str) -> Optional[bool]: """Delete an existing license - API endpoint: PATCH /licenses + API endpoint: DELETE /licenses :param license_shortname: license shortname as the id :type license_shortname: string @@ -76,8 +78,6 @@ def delete_license(self, license_shortname: str) -> Optional[bool]: :raises SW360Error: if there is a negative HTTP response """ - # 2025-01-23: returns 500 - internal server error - if not license_shortname: raise SW360Error(message="No license shortname provided!") @@ -132,7 +132,9 @@ def get_all_licenses(self) -> List[Dict[str, Any]]: :raises SW360Error: if there is a negative HTTP response """ - resp = self.api_get(self.url + "resource/api/licenses") + fullbase_url = self.url + "resource/api/licenses" + resp = self.api_get_all(fullbase_url) + if resp and "_embedded" in resp and "sw360:licenses" in resp["_embedded"]: return resp["_embedded"]["sw360:licenses"] @@ -154,3 +156,26 @@ def get_license(self, license_id: str) -> Optional[Dict[str, Any]]: resp = self.api_get(self.url + "resource/api/licenses/" + license_id) return resp + + def search_license(self, search_text: str) -> Optional[Dict[str, Any]]: + """Get search a license by fullName, shortName or text + + API endpoint: GET /licenses?searchText= + + :param search_text: the search string + :type search_text: string + :return: list of licenses + :rtype: list of JSON license objects + :raises SW360Error: if there is a negative HTTP response + """ + if not search_text: + raise SW360Error(message="Search Text provided!") + + fullbase_url = self.url + "resource/api/licenses" + params = {"searchText": search_text} + full_url = self._add_params(fullbase_url, params) + + sort = LicenseSortColumn.SCORE.asc() + + resp = self.api_get_all(full_url, sort) + return resp diff --git a/sw360/moderationrequests.py b/sw360/moderationrequests.py index 4027d9d..79c2aa7 100644 --- a/sw360/moderationrequests.py +++ b/sw360/moderationrequests.py @@ -12,55 +12,66 @@ from sw360.sw360error import SW360Error from .base import BaseMixin +from .sorting import ModerationSortColumn, SortParam class ModerationRequestMixin(BaseMixin): - def get_all_moderation_requests(self, page: int = -1, page_size: int = -1, - sort: str = "") -> Optional[Dict[str, Any]]: + def get_all_moderation_requests( + self, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Optional[Dict[str, Any]]: """Get information of about all moderation requests API endpoint: GET /moderationrequest :param page: page to retrieve :type page: int - :param page_size: page size to use + :param page_size: page size to use, `-1` to get all :type page_size: int - :param sort: sort order for the packages ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (Sort by request date if + `None`) + :type sort: SortParam :return: list of moderation requests :rtype: list of JSON moderation requests objects :raises SW360Error: if there is a negative HTTP response """ fullbase_url = self.url + "resource/api/moderationrequest" - params = {} + params = {"luceneSearch": "true"} - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) - - if sort: - params["sort"] = sort + if sort is None: + sort = ModerationSortColumn.REQUEST_DATE.desc() full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) + if page > -1 and page_size > -1: + full_url = self._add_pagination(full_url, page, page_size, sort) + + if page_size == -1: + resp = self.api_get_all(full_url, sort) + else: + resp = self.api_get(full_url) + return resp - def get_moderation_requests_by_state(self, state: str, all_details: bool = False, - page: int = -1, page_size: int = -1, - sort: str = "") -> Optional[Dict[str, Any]]: + def get_moderation_requests_by_state( + self, state: str, all_details: bool = False, page: int = -1, + page_size: int = -1, sort: Optional[SortParam] = None + ) -> Optional[Dict[str, Any]]: """Get information of about all moderation requests API endpoint: GET /moderationrequest/byState - :param all_details: retrieve all package details (optional)) + :param state: State of the moderation request. One of "open", "closed" + :type state: string + :param all_details: retrieve all moderation details (optional) :type all_details: bool :param page: page to retrieve :type page: int - :param page_size: page size to use + :param page_size: page size to use, `-1` to get all :type page_size: int - :param sort: sort order for the packages ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (Sort by request date if + `None`) + :type sort: SortParam :return: list of moderation requests :rtype: list of JSON moderation requests objects :raises SW360Error: if there is a negative HTTP response @@ -74,15 +85,18 @@ def get_moderation_requests_by_state(self, state: str, all_details: bool = False if all_details: params["allDetails"] = "true" - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) - - if sort: - params["sort"] = sort + if sort is None: + sort = ModerationSortColumn.REQUEST_DATE.desc() full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) + if page > -1 and page_size > -1: + full_url = self._add_pagination(full_url, page, page_size, sort) + + if page_size == -1: + resp = self.api_get_all(full_url, sort) + else: + resp = self.api_get(full_url) + return resp def get_moderation_request(self, mr_id: str) -> Optional[Dict[str, Any]]: diff --git a/sw360/packages.py b/sw360/packages.py index e86d24f..02f7eee 100644 --- a/sw360/packages.py +++ b/sw360/packages.py @@ -10,10 +10,50 @@ from typing import Any, Dict, List, Optional from .base import BaseMixin +from .sorting import PackageSortColumn, SortParam from .sw360error import SW360Error class PackagesMixin(BaseMixin): + + def __get_packages_filtered( + self, url: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Any: + """ + Take a pre-generated URL of packages endpoint, with filters applied. + Then call the API with appropriate pagination and sorting to get the + packages. + + :param url: Packages API URL with filters in the query + :type url: str + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the package + :type sort: SortParam + :return: list of packages + :rtype: list of JSON package objects + :raises SW360Error: if there is a negative HTTP response + """ + + full_url = self._add_params(url, {"luceneSearch": "true"}) + if page > -1 and page_size > -1: + full_url = self._add_pagination(full_url, page, page_size, sort) + + if page_size == -1: + resp = self.api_get_all(full_url, sort) + else: + resp = self.api_get(full_url) + + if (resp and + "_embedded" in resp and + "sw360:packages" in resp["_embedded"]): + return resp["_embedded"]["sw360:packages"] + + return [] + def get_package(self, package_id: str) -> Optional[Dict[str, Any]]: """Get information of about a package @@ -31,13 +71,22 @@ def get_package(self, package_id: str) -> Optional[Dict[str, Any]]: resp = self.api_get(self.url + "resource/api/packages/" + package_id) return resp - def get_packages_by_name(self, name: str) -> List[Any]: + def get_packages_by_name( + self, name: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Any]: """Gets a list of packages that match the given name. API endpoint: GET /packages?name= :param name: the name :type name: string + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the packages (Sort by score if `None`) + :type sort: SortParam :return: list of packages :rtype: list of JSON package objects :raises SW360Error: if there is a negative HTTP response @@ -45,28 +94,44 @@ def get_packages_by_name(self, name: str) -> List[Any]: if not name: raise SW360Error(message="No package name provided!") - full_url = self.url + "resource/api/packages?name=" + name - resp = self.api_get(full_url) - if resp and ("_embedded" in resp) and ("sw360:packages" in resp["_embedded"]): - return resp["_embedded"]["sw360:packages"] + fullbase_url = self.url + "resource/api/packages" + params = {"name": name} - return [] + url_with_param = self._add_params(fullbase_url, params) + + if sort is None: + sort = PackageSortColumn.SCORE.asc() - def get_all_packages(self, name: str = "", version: str = "", purl: str = "", - all_details: bool = False, page: int = -1, - page_size: int = -1, sort: str = "") -> Any: + return self.__get_packages_filtered(url_with_param, page, page_size, + sort) + + def get_all_packages( + self, name: str = "", version: str = "", purl: str = "", + all_details: bool = False, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Any: """Get information of about all packages API endpoint: GET /releases - :param all_details: retrieve all package details (optional)) + :param name: name to filter + :type name: str + :param version: version to filter + :type version: str + :param purl: purl to filter + :type purl: str + :param all_details: retrieve all package details (optional) :type all_details: bool :param page: page to retrieve :type page: int - :param page_size: page size to use + :param page_size: page size to use, `-1` to get all :type page_size: int - :param sort: sort order for the packages ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (name default, score if + filtering) + :type sort: SortParam + :return: list of packages + :rtype: list of JSON package objects + :raises SW360Error: if there is a negative HTTP response :return: list of packages :rtype: list of JSON package objects :raises SW360Error: if there is a negative HTTP response @@ -86,35 +151,32 @@ def get_all_packages(self, name: str = "", version: str = "", purl: str = "", if purl: params["purl"] = purl - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) - - if sort: - params["sort"] = sort - - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) + url_with_param = self._add_params(fullbase_url, params) - if page == -1 and resp and ("_embedded" in resp) and ("sw360:packages" in resp["_embedded"]): - return resp["_embedded"]["sw360:packages"] + if sort is None: + sort = PackageSortColumn.NAME.asc() + if name != "" or version != "" or purl != "": + sort = PackageSortColumn.SCORE.asc() - return resp + return self.__get_packages_filtered(url_with_param, page, page_size, + sort) - def get_packages_by_packagemanager(self, manager: str, page: int = -1, - page_size: int = -1, sort: str = "") -> Any: + def get_packages_by_packagemanager( + self, manager: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Any: """Get information of about all packages of a specific package manager - API endpoint: GET /releases + API endpoint: GET /releases?packageManager= :param manager: name of the package manager :type manager: str :param page: page to retrieve :type page: int - :param page_size: page size to use + :param page_size: page size to use, `-1` to get all :type page_size: int - :param sort: sort order for the packages ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (Sort by score if `None`) + :type sort: SortParam :return: list of packages :rtype: list of JSON package objects :raises SW360Error: if there is a negative HTTP response @@ -125,20 +187,13 @@ def get_packages_by_packagemanager(self, manager: str, page: int = -1, fullbase_url = self.url + "resource/api/packages" params = {"packageManager": manager} - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) + url_with_param = self._add_params(fullbase_url, params) - if sort: - params["sort"] = sort - - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) - - if page == -1 and resp and ("_embedded" in resp) and ("sw360:packages" in resp["_embedded"]): - return resp["_embedded"]["sw360:packages"] + if sort is None: + sort = PackageSortColumn.SCORE.asc() - return resp + return self.__get_packages_filtered(url_with_param, page, page_size, + sort) def create_new_package(self, name: str, version: str, purl: str, package_type: str, package_details: Dict[str, Any] = {}) -> Optional[Dict[str, Any]]: @@ -185,10 +240,10 @@ def create_new_package(self, name: str, version: str, purl: str, def update_package(self, package: Dict[str, Any], package_id: str) -> Optional[Dict[str, Any]]: """Update an existing package - API endpoint: PATCH /packages + API endpoint: PATCH /packages/{id} - :param release: the new package data - :param release_id: the id of the package to be updated + :param package: the new package data + :param package_id: the id of the package to be updated :type package: JSON :type package_id: string :return: SW360 result @@ -205,7 +260,7 @@ def update_package(self, package: Dict[str, Any], package_id: str) -> Optional[D def delete_package(self, package_id: str) -> Optional[Dict[str, Any]]: """Delete an existing package - API endpoint: DELETE /packages + API endpoint: DELETE /packages/{id} :param package_id: the id of the package to be deleted :type package_id: string @@ -224,3 +279,21 @@ def delete_package(self, package_id: str) -> Optional[Dict[str, Any]]: if response.text: return response.json() return None + + def get_users_of_package(self, package_id: str) -> Optional[Dict[str, Any]]: + """Get information of about the uses of a package + + API endpoint: GET /packages/{id}/usage + + :param package_id: the id of the package to be requested + :type package_id: string + :return: all uses of this package + :rtype: JSON objects + :raises SW360Error: if there is a negative HTTP response + """ + if not package_id: + raise SW360Error(message="No package id provided!") + + resp = self.api_get(self.url + "resource/api/packages/"+ package_id + + "/usage") + return resp diff --git a/sw360/project.py b/sw360/project.py index 6b5d44a..130c77c 100644 --- a/sw360/project.py +++ b/sw360/project.py @@ -12,6 +12,7 @@ from typing import Any, Dict, List, Optional from .base import BaseMixin +from .sorting import ProjectSortColumn, ReleaseSortColumn, SortParam from .sw360error import SW360Error @@ -19,7 +20,7 @@ class ProjectMixin(BaseMixin): def get_project(self, project_id: str) -> Optional[Dict[str, Any]]: """Get information of about a project - API endpoint: GET /projects + API endpoint: GET /projects/{id} :param project_id: the id of the project to be requested :type project_id: string @@ -52,8 +53,9 @@ def get_project_releases(self, project_id: str, transitive: bool = False) -> Any trans = "false" if transitive: trans = "true" - resp = self.api_get(self.url + "resource/api/projects/" - + project_id + "/releases?transitive=" + trans) + url = self.url + "resource/api/projects/" + project_id + "/releases" + url = self._add_params(url, {"transitive": trans}) + resp = self.api_get_all(url, ReleaseSortColumn.CREATED_ON.desc()) return resp def get_project_by_url(self, url: str) -> Optional[Dict[str, Any]]: @@ -73,20 +75,64 @@ def get_project_by_url(self, url: str) -> Optional[Dict[str, Any]]: resp = self.api_get(url) return resp - def get_projects(self, all_details: bool = False, page: int = -1, - page_size: int = -1, sort: str = "") -> Optional[Dict[str, Any]]: + def __get_projects_filtered( + self, url: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: + """ + Take a pre-generated URL of projects endpoint, with filters applied. + Then call the API with appropriate pagination and sorting to get the + projects. + + :param url: projects API URL with filters in the query + :type url: str + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the projects (Sort by name if `None`) + :type sort: SortParam + :return: list of projects + :rtype: list of JSON project objects + :raises SW360Error: if there is a negative HTTP response + """ + + full_url = self._add_params(url, {"luceneSearch": "true"}) + if page > -1 and page_size > -1: + full_url = self._add_pagination(full_url, page, page_size, sort) + + if page_size == -1: + resp = self.api_get_all(full_url, sort) + else: + resp = self.api_get(full_url) + + if not resp: + return [] + + if "_embedded" not in resp: + return [] + + if "sw360:projects" not in resp["_embedded"]: + return [] + + return resp["_embedded"]["sw360:projects"] + + def get_projects( + self, all_details: bool = False, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: """Get all projects API endpoint: GET /projects - :param all_details: retrieve all project details (optional)) + :param all_details: retrieve all project details (optional) :type all_details: bool :param page: page to retrieve :type page: int - :param page_size: page size to use + :param page_size: page size to use, `-1` to get all :type page_size: int - :param sort: sort order for the projects ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the projects (Sort by name if `None`) + :type sort: SortParam :return: list of projects :rtype: list of JSON project objects :raises SW360Error: if there is a negative HTTP response @@ -97,25 +143,30 @@ def get_projects(self, all_details: bool = False, page: int = -1, if all_details: params["allDetails"] = "true" + url_with_param = self._add_params(fullbase_url, params) - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) - - if sort: - params["sort"] = sort + if sort is None: + sort = ProjectSortColumn.NAME.asc() - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) - return resp + return self.__get_projects_filtered(url_with_param, page, page_size, + sort) - def get_projects_by_type(self, project_type: str) -> List[Dict[str, Any]]: + def get_projects_by_type( + self, project_type: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: """Get information of about all projects of a certain type API endpoint: GET /projects :param project_type: the full url of the project to be requested :type project_type: string, one of CUSTOMER, INTERNAL, PRODUCT, SERVICE, INNER_SOURCE + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the projects (Sort by name if `None`) + :type sort: SortParam :return: list of projects :rtype: list of JSON project objects :raises SW360Error: if there is a negative HTTP response @@ -123,17 +174,16 @@ def get_projects_by_type(self, project_type: str) -> List[Dict[str, Any]]: if not project_type: raise SW360Error(message="No project type provided!") - resp = self.api_get(self.url + "resource/api/projects?type=" + project_type) - if not resp: - return [] + fullbase_url = self.url + "resource/api/projects" + params = {"type": project_type} - if "_embedded" not in resp: - return [] + url_with_param = self._add_params(fullbase_url, params) - if "sw360:projects" not in resp["_embedded"]: - return [] + if sort is None: + sort = ProjectSortColumn.NAME.asc() - return resp["_embedded"]["sw360:projects"] + return self.__get_projects_filtered(url_with_param, page, page_size, + sort) def get_project_names(self) -> List[str]: """Get all project names @@ -149,13 +199,7 @@ def get_project_names(self) -> List[str]: if not all_projects: return resp - if "_embedded" not in all_projects: - return resp - - if "sw360:projects" not in all_projects["_embedded"]: - return resp - - projects: List[Dict[str, Any]] = all_projects["_embedded"]["sw360:projects"] + projects: List[Dict[str, Any]] = all_projects if not projects: return resp @@ -164,13 +208,22 @@ def get_project_names(self) -> List[str]: return resp - def get_projects_by_name(self, name: str) -> List[Dict[str, Any]]: + def get_projects_by_name( + self, name: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: """Get a project by its name API endpoint: GET /projects :param name: the project name or a prefix of it :type name: string + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the projects (Sort by score if `None`) + :type sort: SortParam :return: list of projects :rtype: list of JSON project objects :raises SW360Error: if there is a negative HTTP response @@ -178,19 +231,20 @@ def get_projects_by_name(self, name: str) -> List[Dict[str, Any]]: if not name: raise SW360Error(message="No project name provided!") - resp = self.api_get(self.url + "resource/api/projects?name=" + name) - if not resp: - return [] + fullbase_url = self.url + "resource/api/projects" + params = {"name": name} - if "_embedded" not in resp: - return [] + url_with_param = self._add_params(fullbase_url, params) - if "sw360:projects" not in resp["_embedded"]: - return [] + if sort is None: + sort = ProjectSortColumn.SCORE.asc() - return resp["_embedded"]["sw360:projects"] + return self.__get_projects_filtered(url_with_param, page, page_size, + sort) - def get_projects_by_external_id(self, ext_id_name: str, ext_id_value: str = "") -> List[Dict[str, Any]]: + def get_projects_by_external_id( + self, ext_id_name: str, ext_id_value: str = "" + ) -> List[Dict[str, Any]]: """Get projects by external id. `ext_id_value` can be left blank to search for all projects with `ext_id_name`. @@ -223,13 +277,24 @@ def get_projects_by_external_id(self, ext_id_name: str, ext_id_value: str = "") return resp["_embedded"]["sw360:projects"] - def get_projects_by_group(self, group: str, all_details: bool = False) -> List[Dict[str, Any]]: + def get_projects_by_group( + self, group: str, all_details: bool = False, page: int = -1, + page_size: int = -1, sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: """Get projects by group. API endpoint: GET /projects?group= :param group: the group the projects shall belong to :type group: string + :param all_details: retrieve all project details (optional) + :type all_details: bool + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the projects (Sort by score if `None`) + :type sort: SortParam :return: list of projects :rtype: list of JSON project objects :raises SW360Error: if there is a negative HTTP response @@ -237,29 +302,36 @@ def get_projects_by_group(self, group: str, all_details: bool = False) -> List[D if not group: raise SW360Error(message="No group provided!") - full_url = self.url + "resource/api/projects?group=" + group - if all_details: - full_url = self.url + "resource/api/projects?allDetails?group=" + group + fullbase_url = self.url + "resource/api/projects" + params = {"group": group} - resp = self.api_get(full_url) - if not resp: - return [] + if all_details: + params["allDetails"] = "true" - if "_embedded" not in resp: - return [] + url_with_param = self._add_params(fullbase_url, params) - if "sw360:projects" not in resp["_embedded"]: - return [] + if sort is None: + sort = ProjectSortColumn.SCORE.asc() - return resp["_embedded"]["sw360:projects"] + return self.__get_projects_filtered(url_with_param, page, page_size, + sort) - def get_projects_by_tag(self, tag: str) -> List[Dict[str, Any]]: + def get_projects_by_tag( + self, tag: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: """Get projects by tag. API endpoint: GET /projects?tag= - :param group: the group the projects shall belong to - :type group: string + :param tag: the group the projects shall belong to + :type tag: string + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the projects (Sort by score if `None`) + :type sort: SortParam :return: list of projects :rtype: list of JSON project objects :raises SW360Error: if there is a negative HTTP response @@ -267,18 +339,16 @@ def get_projects_by_tag(self, tag: str) -> List[Dict[str, Any]]: if not tag: raise SW360Error(message="No tag provided!") - full_url = self.url + "resource/api/projects?tag=" + tag + "&luceneSearch=true" - resp = self.api_get(full_url) - if not resp: - return [] + fullbase_url = self.url + "resource/api/projects" + params = {"tag": tag} - if "_embedded" not in resp: - return [] + url_with_param = self._add_params(fullbase_url, params) - if "sw360:projects" not in resp["_embedded"]: - return [] + if sort is None: + sort = ProjectSortColumn.SCORE.asc() - return resp["_embedded"]["sw360:projects"] + return self.__get_projects_filtered(url_with_param, page, page_size, + sort) def get_project_vulnerabilities(self, project_id: str) -> Optional[Dict[str, Any]]: """Get the security vulnerabilities for the specified project. @@ -295,15 +365,17 @@ def get_project_vulnerabilities(self, project_id: str) -> Optional[Dict[str, Any raise SW360Error(message="No project id provided!") full_url = self.url + "resource/api/projects/" + project_id + "/vulnerabilities" - resp = self.api_get(full_url) + resp = self.api_get_all(full_url) if not resp: return None return resp - def create_new_project(self, name: str, project_type: str, visibility: Any, - description: str = "", version: str = "", - project_details: Dict[str, Any] = {}) -> Optional[Dict[str, Any]]: + def create_new_project( + self, name: str, project_type: str, visibility: Any, + description: str = "", version: str = "", + project_details: Dict[str, Any] = {} + ) -> Optional[Dict[str, Any]]: """Create a new project. The parameters list only the most common project attributes, check the @@ -341,8 +413,10 @@ def create_new_project(self, name: str, project_type: str, visibility: Any, return response.json() raise SW360Error(response, url) - def update_project(self, project: Dict[str, Any], project_id: str, - add_subprojects: bool = False) -> Optional[Dict[str, Any]]: + def update_project( + self, project: Dict[str, Any], project_id: str, + add_subprojects: bool = False + ) -> Optional[Dict[str, Any]]: """Update an existing project API endpoint: PATCH /projects @@ -367,12 +441,15 @@ def update_project(self, project: Dict[str, Any], project_id: str, if add_subprojects: current = self.get_project(project_id) - if (current is not None and "linkedProjects" in current and "linkedProjects" in project): + if current is not None and "linkedProjects" in current and "linkedProjects" in project: for sp in current["linkedProjects"]: pid = self.get_id_from_href(sp["project"]) if pid not in project["linkedProjects"]: - nsp = {} - nsp["projectRelationship"] = sp.get("relation", "CONTAINED") + nsp = { + "projectRelationship": sp.get( + "relation", "CONTAINED" + ) + } project["linkedProjects"][pid] = nsp return self.api_patch(url, json=project) @@ -385,13 +462,13 @@ def update_project_releases( given `releases` are added to the project, otherwise, the existing releases will be replaced. - API endpoint: POST /projects//releases + API endpoint: POST /projects/{id}/releases - :param releases: list of relase_ids to be linked in the project - :param project_id: the id of the project to modify - :param add: add given releases if set to True, replace otherwise + :param releases: list of release_ids to be linked in the project :type releases: list of release_id strings + :param project_id: the id of the project to modify :type project_id: string + :param add: add given releases if set to True, replace otherwise :type add: boolean :return: SW360 result :rtype: JSON SW360 result object @@ -421,7 +498,7 @@ def update_project_external_id(self, ext_id_name: str, ext_id_value: str, The method will return the old value of the external id or None if it was not set. - API endpoint: PATCH /projects + API endpoint: PATCH /projects/{id} :param ext_id_name: name of the external id :param ext_id_value: value of the external id @@ -458,7 +535,7 @@ def update_project_external_id(self, ext_id_name: str, ext_id_value: str, def delete_project(self, project_id: str) -> Optional[Dict[str, Any]]: """Delete an existing project - API endpoint: DELETE /projects + API endpoint: DELETE /projects/{id} :param project_id: the id of the project to be requested :type project_id: string @@ -499,7 +576,7 @@ def get_users_of_project(self, project_id: str) -> Optional[Dict[str, Any]]: def duplicate_project(self, project_id: str, new_version: str) -> Optional[Dict[str, Any]]: """Create a copy of an existing project. - API endpoint: GET /projects/duplicate/{id} + API endpoint: POST /projects/duplicate/{id} :param project_id: the id of the existing project :type project_id: string @@ -534,7 +611,7 @@ def update_project_release_relationship( new_relation: str, comment: str) -> Optional[Dict[str, Any]]: """Update the relationship for a specific release of a project - API endpoint PATCH /projects/{pid}/release{rid} + API endpoint PATCH /projects/{pid}/release/{rid} :param project_id: the id of the existing project :type project_id: string @@ -573,7 +650,7 @@ def update_project_release_relationship( def link_packages_to_project(self, project_id: str, packages: List[str]) -> Optional[Dict[str, Any]]: """Link (new) packages to a given project. - API endpoint PATCH /projects/{pid}/packages{rid} + API endpoint PATCH /projects/{pid}/packages/{rid} :param project_id: the id of the existing project :type project_id: string @@ -594,7 +671,7 @@ def link_packages_to_project(self, project_id: str, packages: List[str]) -> Opti def unlink_packages_from_project(self, project_id: str, packages: List[str]) -> Optional[Dict[str, Any]]: """Unlink packages from a given project. - API endpoint PATCH /projects/{pid}/packages{rid} + API endpoint PATCH /projects/{pid}/packages/{rid} :param project_id: the id of the existing project :type project_id: string @@ -611,3 +688,35 @@ def unlink_packages_from_project(self, project_id: str, packages: List[str]) -> url = self.url + "resource/api/projects/" + project_id + "/unlink/packages/" return self.api_patch(url, json=packages) + + def upload_attachment_to_project( + self, project_id: str, upload_file: str, upload_type: str = "SOURCE", + upload_comment: str = "" + ) -> Optional[Dict[str, Any]]: + """Upload an attachment to a given project. + + API endpoint: POST /attachments & PATCH /projects/{id} + + :param project_id: the id of the Project + :type project_id: string + :param upload_file: path of the file to be uploaded + :type upload_file: string + :param upload_type: the type of the attachment + :type upload_type: string + :param upload_comment: a comment for the attachment + :type upload_comment: string + :raises SW360Error: if the project id is missing or there is a negative HTTP response + """ + if not project_id: + raise SW360Error(message="No project id provided!") + + attachment_content = self._upload_resource_file(upload_file, upload_type, upload_comment) + # Make sure the type is correct, override + attachment_content['attachmentType'] = upload_type + attachment_content['createdComment'] = upload_comment + + current_project = self.get_project(project_id) + attachments = self._get_attachments(current_project) + + attachments.append(attachment_content) + return self.update_project({'attachments': attachments}, project_id) diff --git a/sw360/releases.py b/sw360/releases.py index 3ad5371..5bed6cd 100644 --- a/sw360/releases.py +++ b/sw360/releases.py @@ -12,10 +12,50 @@ from typing import Any, Dict, List, Optional from .base import BaseMixin +from .sorting import ReleaseSortColumn, SortParam from .sw360error import SW360Error class ReleasesMixin(BaseMixin): + + def __get_releases_filtered( + self, url: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Any: + """ + Take a pre-generated URL of releases endpoint, with filters applied. + Then call the API with appropriate pagination and sorting to get the + releases. + + :param url: Releases API URL with filters in the query + :type url: str + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the releases + :type sort: SortParam + :return: list of releases + :rtype: list of JSON release objects + :raises SW360Error: if there is a negative HTTP response + """ + + full_url = self._add_params(url, {"luceneSearch": "true"}) + if page > -1 and page_size > -1: + full_url = self._add_pagination(full_url, page, page_size, sort) + + if page_size == -1: + resp = self.api_get_all(full_url, sort) + else: + resp = self.api_get(full_url) + + if (resp and + "_embedded" in resp and + "sw360:releases" in resp["_embedded"]): + return resp["_embedded"]["sw360:releases"] + + return [] + def get_release(self, release_id: str) -> Optional[Dict[str, Any]]: """Get information of about a release @@ -38,8 +78,8 @@ def get_release_by_url(self, release_url: str) -> Optional[Dict[str, Any]]: API endpoint: GET /releases - :param url: the full url of the release to be requested - :type url: string + :param release_url: the full url of the release to be requested + :type release_url: string :return: a release :rtype: JSON release object :raises SW360Error: if there is a negative HTTP response @@ -50,13 +90,22 @@ def get_release_by_url(self, release_url: str) -> Optional[Dict[str, Any]]: resp = self.api_get(release_url) return resp - def get_releases_by_name(self, name: str) -> List[Any]: + def get_releases_by_name( + self, name: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Any]: """Gets a list of releases that match the given name. API endpoint: GET /releases?name= :param name: the name :type name: string + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the releases (Sort by score if `None`) + :type sort: SortParam :return: list of releases :rtype: list of JSON release objects :raises SW360Error: if there is a negative HTTP response @@ -64,32 +113,40 @@ def get_releases_by_name(self, name: str) -> List[Any]: if not name: raise SW360Error(message="No release name provided!") - full_url = self.url + "resource/api/releases?name=" + name - resp = self.api_get(full_url) - if resp and ("_embedded" in resp) and ("sw360:releases" in resp["_embedded"]): - return resp["_embedded"]["sw360:releases"] + fullbase_url = self.url + "resource/api/releases" + params = {"name": name} - return [] + url_with_param = self._add_params(fullbase_url, params) + + if sort is None: + sort = ReleaseSortColumn.SCORE.asc() + + return self.__get_releases_filtered(url_with_param, page, page_size, + sort) # return type List[Dict[str, Any]] | Optional[Dict[str, Any]] for Python 3.11 is good, # Union[List[Dict[str, Any]], Optional[Dict[str, Any]]] for lower Python versions is not good - def get_all_releases(self, fields: str = "", all_details: bool = False, - isNewClearingWithSourceAvailable: bool = False, page: int = -1, - page_size: int = -1, sort: str = "") -> Any: + def get_all_releases( + self, fields: str = "", all_details: bool = False, + isNewClearingWithSourceAvailable: bool = False, page: int = -1, + page_size: int = -1, sort: Optional[SortParam] = None + ) -> Any: """Get information of about all releases API endpoint: GET /releases - :param all_details: retrieve all project details (optional)) + :param fields: fields to include + :type fields: string + :param all_details: retrieve all release details (optional) :type all_details: bool - :param isNewClearingWithSourceAvailable: retrieve releases in new clearning state with source avail + :param isNewClearingWithSourceAvailable: retrieve releases in new clearing state with source available :type isNewClearingWithSourceAvailable: bool :param page: page to retrieve :type page: int - :param page_size: page size to use + :param page_size: page size to use, `-1` to get all :type page_size: int - :param sort: sort order for the releases ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the releases (Sort by name if `None`) + :type sort: SortParam :return: list of releases :rtype: list of JSON release objects :raises SW360Error: if there is a negative HTTP response @@ -106,31 +163,23 @@ def get_all_releases(self, fields: str = "", all_details: bool = False, if fields: params["fields"] = fields - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) - - if sort: - params["sort"] = sort - - full_url = self._add_params(fullbase_url, params) - - resp = self.api_get(full_url) + url_with_param = self._add_params(fullbase_url, params) - if page == -1 and resp and ("_embedded" in resp) and ("sw360:releases" in resp["_embedded"]): - return resp["_embedded"]["sw360:releases"] + if sort is None: + sort = ReleaseSortColumn.NAME.asc() - return resp + return self.__get_releases_filtered(url_with_param, page, page_size, + sort) def get_releases_by_external_id(self, ext_id_name: str, ext_id_value: str = "") -> List[Dict[str, Any]]: """Get releases by external id. `ext_id_value` can be left blank to search for all releases with `ext_id_name`. - API endpoint: GET /releases + API endpoint: GET /releases/searchByExternalIds :param ext_id_name: the name of the external id to look for - :param ext_id_value: the value of the external id to look for :type ext_id_name: string + :param ext_id_value: the value of the external id to look for :type ext_id_value: string :return: list of releases :rtype: list of JSON release objects @@ -139,11 +188,14 @@ def get_releases_by_external_id(self, ext_id_name: str, ext_id_value: str = "") if not ext_id_name: raise SW360Error(message="No external id name provided!") - resp = self.api_get( - self.url - + "resource/api/releases/searchByExternalIds?" - + ext_id_name + "=" + ext_id_value + url = self._add_params( + self.url + "resource/api/releases/searchByExternalIds", + { + ext_id_name: ext_id_value + } ) + + resp = self.api_get(url) if resp and ("_embedded" in resp) and ("sw360:releases" in resp["_embedded"]): return resp["_embedded"]["sw360:releases"] @@ -291,7 +343,7 @@ def get_users_of_release(self, release_id: str) -> Optional[Dict[str, Any]]: def link_packages_to_release(self, release_id: str, packages: List[str]) -> Optional[Dict[str, Any]]: """Link (new) packages to a given release. - API endpoint PATCH /release/{pid}/packages{rid} + API endpoint PATCH /release/{rid}/link/packages :param release_id: the id of the existing release :type release_id: string @@ -345,3 +397,33 @@ def get_recent_releases(self) -> Optional[List[Dict[str, Any]]]: return resp["_embedded"]["sw360:releases"] return [] + + def upload_attachment_to_release( + self, release_id: str, upload_file: str, upload_type: str = "SOURCE", + upload_comment: str = "" + ) -> Optional[Dict[str, Any]]: + """Upload an attachment to a given Release. + + API endpoint: POST /attachments & PATCH /releases/{id} + + :param release_id: the id of the Release + :type release_id: string + :param upload_file: path of the file to be uploaded + :type upload_file: string + :param upload_type: the type of the attachment + :type upload_type: string + :param upload_comment: a comment for the attachment + :type upload_comment: string + :raises SW360Error: if the release id is missing or there is a negative HTTP response + """ + if not release_id: + raise SW360Error(message="No release id provided!") + + attachment_content = self._upload_resource_file(upload_file, upload_type, upload_comment) + attachment_content['attachmentType'] = upload_type # Make sure the type is correct + attachment_content['createdComment'] = upload_comment # Override + + current_release = self.get_release(release_id) + attachments = self._get_attachments(current_release) + attachments.append(attachment_content) + return self.update_release({'attachments': attachments}, release_id) diff --git a/sw360/sorting.py b/sw360/sorting.py new file mode 100644 index 0000000..a8358a4 --- /dev/null +++ b/sw360/sorting.py @@ -0,0 +1,88 @@ +# ------------------------------------------------------------------------------ +# Copyright (c) 2026 Siemens +# All Rights Reserved. +# Authors: mishra.gaurav@siemens.com +# +# Licensed under the MIT license. +# SPDX-License-Identifier: MIT +# ------------------------------------------------------------------------------ +from enum import Enum + + +class SortParam: + def __init__(self, value: str): + self.value = value + + def __str__(self) -> str: + return self.value + + +class BaseSortMixin(Enum): + def asc(self) -> SortParam: + return SortParam(f"{self.value},asc") + + def desc(self) -> SortParam: + return SortParam(f"{self.value},desc") + + +class ProjectSortColumn(BaseSortMixin, Enum): + SCORE = "score" + CREATED_ON = "createdOn" + NAME = "name" + VENDOR = "vendor" + LICENSE = "license" + TYPE = "type" + DESCRIPTION = "description" + PROJECT_RESPONSIBLE = "projectResponsible" + STATE = "state" + + +class ReleaseSortColumn(BaseSortMixin, Enum): + CREATED_ON = "createdOn" + NAME = "name" + VERSION = "version" + CLEARING_STATE = "clearingState" + MAINLINE_STATE = "mainlineState" + SCORE = "score" + +class ComponentSortColumn(BaseSortMixin, Enum): + SCORE = "score" + CREATED_ON = "createdOn" + NAME = "name" + VENDOR_NAMES = "vendorNames" + MAIN_LICENSE_IDS = "mainLicenseIds" + TYPE = "type" + +class LicenseSortColumn(BaseSortMixin, Enum): + SCORE = "score" + FULL_NAME = "fullName" + SHORT_NAME = "shortName" + +class ModerationSortColumn(BaseSortMixin, Enum): + SCORE = "score" + DOCUMENT_NAME = "documentName" + DOCUMENT_TYPE = "documentType" + COMPONENT_TYPE = "componentType" + MODERATION_STATE = "moderationState" + REQUEST_DATE = "requestDate" + REQUESTING_USER = "requestingUser" + REQUESTING_USER_DEPARTMENT = "requestingUserDepartment" + +class PackageSortColumn(BaseSortMixin, Enum): + SCORE = "score" + CREATED_ON = "createdOn" + NAME = "name" + VERSION = "version" + PACKAGE_MANAGER = "packageManager" + +class VendorSortColumn(BaseSortMixin, Enum): + SCORE = "score" + FULL_NAME = "fullName" + SHORT_NAME = "shortName" + +class VulnerabilitySortColumn(BaseSortMixin, Enum): + LAST_UPDATE_DATE = "lastUpdateDate" + EXTERNAL_ID = "externalId" + TITLE = "title" + CVSS = "cvss" + PUBLISH_DATE = "publishDate" diff --git a/sw360/sw360_api.py b/sw360/sw360_api.py index 795f4cf..d6450ce 100644 --- a/sw360/sw360_api.py +++ b/sw360/sw360_api.py @@ -68,6 +68,8 @@ class SW360( :type url: string :type token: string :type oauth2: boolean + :ivar default_batch_size: Default size of batch to use while fetching all items from API + :type default_batch_size: int """ def __init__( @@ -75,9 +77,11 @@ def __init__( url: str, token: str, oauth2: bool = False, - session: Optional[requests.Session] = session_default + session: Optional[requests.Session] = session_default, + default_batch_size: int = 50 ) -> None: """Constructor""" + super().__init__(url, token, oauth2, default_batch_size) if url[-1] != "/": url += "/" self.url: str = url @@ -89,6 +93,7 @@ def __init__( self.api_headers = {"Authorization": "Token " + token} self.force_no_session = False + self.default_batch_size = default_batch_size def login_api(self, token: str = "") -> bool: """Login to SW360 REST API. This used to have a `token` parameter diff --git a/sw360/vendor.py b/sw360/vendor.py index 0d25109..8906cc8 100644 --- a/sw360/vendor.py +++ b/sw360/vendor.py @@ -12,21 +12,86 @@ from typing import Any, Dict, List, Optional from .base import BaseMixin +from .sorting import SortParam, VendorSortColumn from .sw360error import SW360Error class VendorMixin(BaseMixin): - def get_all_vendors(self) -> List[Dict[str, Any]]: + def get_all_vendors( + self, page: int = -1, page_size: int = -1, sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: """Returns all vendors API endpoint: GET /vendors - :return: a vendor - :rtype: JSON vendor object + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the vendors (Sort by name if `None`) + :type sort: SortParam + :return: list of vendors + :rtype: list of JSON vendor objects :raises SW360Error: if there is a negative HTTP response """ - resp = self.api_get(self.url + "resource/api/vendors") + fullbase_url = self.url + "resource/api/vendors" + + if sort is None: + sort = VendorSortColumn.SHORT_NAME.asc() + + full_url = fullbase_url + if page > -1 and page_size > -1: + full_url = self._add_pagination(full_url, page, page_size, sort) + + if page_size == -1: + resp = self.api_get_all(full_url, sort) + else: + resp = self.api_get(full_url) + + if resp and "_embedded" in resp and "sw360:vendors" in resp["_embedded"]: + return resp["_embedded"]["sw360:vendors"] + + return [] + + def search_vendors( + self, search_text: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: + """Search vendors by full name or short name + + API endpoint: GET /vendors?searchText={search_text} + + :param search_text: search text + :type search_text: string + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the vendors (Sort by score if `None`) + :type sort: SortParam + :return: list of vendors + :rtype: list of JSON vendor objects + :raises SW360Error: if there is a negative HTTP response + """ + if not search_text: + raise SW360Error(message="No search text provided!") + + fullbase_url = self.url + "resource/api/vendors" + params = {"searchText": search_text} + + if sort is None: + sort = VendorSortColumn.SCORE.asc() + + full_url = self._add_params(fullbase_url, params) + if page > -1 and page_size > -1: + full_url = self._add_pagination(full_url, page, page_size, sort) + + if page_size == -1: + resp = self.api_get_all(full_url, sort) + else: + resp = self.api_get(full_url) + if resp and "_embedded" in resp and "sw360:vendors" in resp["_embedded"]: return resp["_embedded"]["sw360:vendors"] @@ -72,15 +137,15 @@ def create_new_vendor(self, vendor: Dict[str, Any]) -> Dict[str, Any]: def update_vendor(self, vendor: Dict[str, Any], vendor_id: str) -> Optional[Dict[str, Any]]: """Update an existing vendor - API endpoint: PATCH /vendors + API endpoint: PATCH /vendors/{id} + :param vendor: the new vendor data + :type vendor: JSON vendor object :param vendor_id: the id of the vendor to be updated :type vendor_id: string :raises SW360Error: if there is a negative HTTP response """ - # 2019-04-03: error 405 - not allowed - if not vendor_id: raise SW360Error(message="No vendor id provided!") @@ -90,15 +155,13 @@ def update_vendor(self, vendor: Dict[str, Any], vendor_id: str) -> Optional[Dict def delete_vendor(self, vendor_id: str) -> Optional[Dict[str, Any]]: """Delete an existing vendor - API endpoint: DELETE /vendors + API endpoint: DELETE /vendors/{id} :param vendor_id: the id of the vendor :type vendor_id: string :raises SW360Error: if there is a negative HTTP response """ - # 2019-04-03/2025-01-23: error 405 - not allowed - if not vendor_id: raise SW360Error(message="No vendor id provided!") @@ -111,3 +174,20 @@ def delete_vendor(self, vendor_id: str) -> Optional[Dict[str, Any]]: return response.json() return None + + def get_users_of_vendor(self, vendor_id: str) -> Optional[Dict[str, Any]]: + """Get information of about the releases supplied by the vendor + + API endpoint: GET /vendors/{id}/releases + + :param vendor_id: the id of the vendor to be requested + :type vendor_id: string + :return: all releases supplied by this vendor + :rtype: JSON objects + :raises SW360Error: if there is a negative HTTP response + """ + if not vendor_id: + raise SW360Error(message="No vendor id provided!") + + resp = self.api_get(self.url + "resource/api/vendors/" + vendor_id + "/releases") + return resp diff --git a/sw360/vulnerabilities.py b/sw360/vulnerabilities.py index 33516f8..616ba98 100644 --- a/sw360/vulnerabilities.py +++ b/sw360/vulnerabilities.py @@ -12,20 +12,56 @@ from typing import Any, Dict, Optional from .base import BaseMixin +from .sorting import SortParam, VulnerabilitySortColumn class VulnerabilitiesMixin(BaseMixin): - def get_all_vulnerabilities(self) -> Optional[Dict[str, Any]]: + def get_all_vulnerabilities( + self, search_text: str = "", all_details: bool = False, + page: int = 0, page_size: int = 10, sort: Optional[SortParam] = None + ) -> Optional[Dict[str, Any]]: """Get information of about all vulnerabilities API endpoint: GET /vulnerabilities + :param search_text: filter by externalId or title of vulnerability, + empty to get all + :type search_text: string + :param all_details: return all details about vulnerabilities ( + including releases) + :type all_details: bool + :param page: page to retrieve + :type page: int + :param page_size: page size to use + :type page_size: int + :param sort: sort order for the vulnerabilities (Sort by last update if + `None`) + :type sort: SortParam :return: list of vulnerabilities :rtype: list of JSON vulnerability objects :raises SW360Error: if there is a negative HTTP response """ + fullbase_url = self.url + "resource/api/vulnerabilities" + params = {} + + if search_text != "": + params["search"] = search_text + + if all_details: + params["allDetails"] = "true" + + if sort is None: + sort = VulnerabilitySortColumn.LAST_UPDATE_DATE.desc() + + full_url = self._add_params(fullbase_url, params) + if page > -1 and page_size > -1: + full_url = self._add_pagination(full_url, page, page_size, sort) + + if page_size == -1: + resp = self.api_get_all(full_url, sort) + else: + resp = self.api_get(full_url) - resp = self.api_get(self.url + "resource/api/vulnerabilities") return resp def get_vulnerability(self, vulnerability_id: str) -> Optional[Dict[str, Any]]: diff --git a/tests/test_sw360_components.py b/tests/test_sw360_components.py index 8d0d209..029a1cf 100644 --- a/tests/test_sw360_components.py +++ b/tests/test_sw360_components.py @@ -18,6 +18,7 @@ sys.path.insert(1, "..") from sw360 import SW360, SW360Error # noqa: E402 +from sw360.sorting import ComponentSortColumn # noqa: E402 class Sw360TestComponents(unittest.TestCase): @@ -105,7 +106,7 @@ def test_get_all_components_with_fields(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?fields=ownerCountry", # noqa + url=self.MYURL + "resource/api/components?fields=ownerCountry&luceneSearch=true&page=0&page_entries=50&sort=name,asc", # noqa body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", @@ -127,15 +128,14 @@ def test_get_all_components_with_fields_and_paging(self) -> None: self.assertTrue(actual) responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?fields=ownerCountry&page=1&page_entries=2", # noqa + url=self.MYURL + "resource/api/components?fields=ownerCountry&luceneSearch=true&page=1&page_entries=2&sort=name,asc", # noqa body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - data = lib.get_all_components("ownerCountry", 1, 2) - components = data["_embedded"]["sw360:components"] + components = lib.get_all_components("ownerCountry", 1, 2) self.assertIsNotNone(components) self.assertTrue(len(components) > 0) self.assertEqual("Tethys.Logging", components[0]["name"]) @@ -151,7 +151,7 @@ def test_get_all_components_with_all_details(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?allDetails=true", # noqa + url=self.MYURL + "resource/api/components?allDetails=true&luceneSearch=true&page=0&page_entries=50&sort=name,asc", # noqa body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", @@ -174,14 +174,14 @@ def test_get_all_components_with_all_details_and_sorting(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?allDetails=true&sort=name%2Cdesc", # noqa + url=self.MYURL + "resource/api/components?allDetails=true&luceneSearch=true&page=0&page_entries=50&sort=name,desc", # noqa body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - components = lib.get_all_components(all_details=True, sort="name,desc") + components = lib.get_all_components(all_details=True, sort=ComponentSortColumn.NAME.desc()) self.assertIsNotNone(components) self.assertTrue(len(components) > 0) self.assertEqual("Tethys.Logging", components[0]["name"]) @@ -204,7 +204,7 @@ def test_get_all_components_invalid_reply(self) -> None: adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - components = lib.get_all_components(all_details=True, sort="name,desc") + components = lib.get_all_components(all_details=True, sort=ComponentSortColumn.NAME.desc()) self.assertIsNotNone(components) self.assertTrue(len(components) == 0) @@ -225,7 +225,7 @@ def test_get_all_components_invalid_reply2(self) -> None: adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - components = lib.get_all_components(all_details=True, sort="name,desc") + components = lib.get_all_components(all_details=True, sort=ComponentSortColumn.NAME.desc()) self.assertIsNotNone(components) self.assertTrue(len(components) == 0) @@ -239,7 +239,7 @@ def test_get_all_components_by_type(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?type=OSS", + url=self.MYURL + "resource/api/components?type=OSS&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", @@ -262,7 +262,7 @@ def test_get_all_components_by_type_no_result(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?type=OSS", + url=self.MYURL + "resource/api/components?type=OSS&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{}', status=200, content_type="application/json", @@ -282,7 +282,7 @@ def test_get_all_components_by_type_invalid_reply(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?type=OSS", + url=self.MYURL + "resource/api/components?type=OSS&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_xxembedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", @@ -303,7 +303,7 @@ def test_get_all_components_by_type_invalid_reply2(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?type=OSS", + url=self.MYURL + "resource/api/components?type=OSS&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"xxsw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", @@ -366,7 +366,7 @@ def test_get_component_by_name(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?name=MyComponent", + url=self.MYURL + "resource/api/components?name=MyComponent&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"name": "MyComponent"}', status=200, content_type="application/json", diff --git a/tests/test_sw360_moderationrequests.py b/tests/test_sw360_moderationrequests.py index b53f472..44b4cbd 100644 --- a/tests/test_sw360_moderationrequests.py +++ b/tests/test_sw360_moderationrequests.py @@ -14,6 +14,7 @@ import responses from sw360 import SW360 +from sw360.sorting import ModerationSortColumn sys.path.insert(1, "..") @@ -50,7 +51,7 @@ def test_get_all_moderation_requests(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/moderationrequest?page=2&page_entries=8&sort=timestamp%2Cdesc", + url=self.MYURL + "resource/api/moderationrequest?luceneSearch=true&page=2&page_entries=8&sort=requestDate%2Cdesc", body='''{ "_embedded": { "sw360:moderationRequests": [ @@ -104,7 +105,7 @@ def test_get_all_moderation_requests(self) -> None: adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - mrs = lib.get_all_moderation_requests(page=2, page_size=8, sort="timestamp,desc") + mrs = lib.get_all_moderation_requests(page=2, page_size=8, sort=ModerationSortColumn.REQUEST_DATE.desc()) self.assertIsNotNone(mrs) if mrs: self.assertTrue(len(mrs) > 0) @@ -120,7 +121,7 @@ def test_get_moderation_requests_by_state(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/moderationrequest/byState?state=open&allDetails=true&page=2&page_entries=8&sort=timestamp%2Cdesc", # noqa + url=self.MYURL + "resource/api/moderationrequest/byState?state=open&allDetails=true&page=2&page_entries=8&sort=requestDate%2Cdesc", # noqa body='''{ "_embedded": { "sw360:moderationRequests": [ @@ -174,7 +175,7 @@ def test_get_moderation_requests_by_state(self) -> None: adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - mrs = lib.get_moderation_requests_by_state("open", True, page=2, page_size=8, sort="timestamp,desc") + mrs = lib.get_moderation_requests_by_state("open", True, page=2, page_size=8, sort=ModerationSortColumn.REQUEST_DATE.desc()) self.assertIsNotNone(mrs) if mrs: self.assertTrue(len(mrs) > 0) diff --git a/tests/test_sw360_packages.py b/tests/test_sw360_packages.py index 4c067b3..f692a72 100644 --- a/tests/test_sw360_packages.py +++ b/tests/test_sw360_packages.py @@ -17,6 +17,7 @@ sys.path.insert(1, "..") from sw360 import SW360, SW360Error # noqa: E402 +from sw360.sorting import PackageSortColumn # noqa: E402 class Sw360TestPackages(unittest.TestCase): @@ -106,7 +107,7 @@ def test_get_all_packages_with_fields_and_all_details(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/packages?allDetails=true&purl=pkg:pypi/cli-support@2.0.0", + url=self.MYURL + "resource/api/packages?allDetails=true&purl=pkg:pypi/cli-support@2.0.0&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"sw360:packages": [{"name": "Tethys.Logging", "version": "1.3.0", "packageType": "FRAMEWORK"}]}}', # noqa status=200, content_type="application/json", @@ -128,19 +129,18 @@ def test_get_all_packages_with_fields_and_all_details2(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/packages?allDetails=true&name=cli-support&version=2.0.0&page=2&page_entries=6&sort=name%2Cdesc", # noqa + url=self.MYURL + "resource/api/packages?allDetails=true&name=cli-support&version=2.0.0&luceneSearch=true&page=2&page_entries=6&sort=name,desc", # noqa body='{"_embedded": {"sw360:packages": [{"name": "Tethys.Logging", "version": "1.3.0", "packageType": "FRAMEWORK"}]}}', # noqa status=200, content_type="application/json", adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - packages = lib.get_all_packages(name="cli-support", version="2.0.0", all_details=True, page=2, page_size=6, sort="name,desc") # noqa + packages = lib.get_all_packages(name="cli-support", version="2.0.0", all_details=True, page=2, page_size=6, sort=PackageSortColumn.NAME.desc()) # noqa self.assertIsNotNone(packages) self.assertTrue(len(packages) > 0) - pkgs = packages["_embedded"]["sw360:packages"] - self.assertEqual("Tethys.Logging", pkgs[0]["name"]) - self.assertEqual("1.3.0", pkgs[0]["version"]) + self.assertEqual("Tethys.Logging", packages[0]["name"]) + self.assertEqual("1.3.0", packages[0]["version"]) @responses.activate def test_get_packages_by_name(self) -> None: @@ -151,7 +151,7 @@ def test_get_packages_by_name(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/packages?name=john", + url=self.MYURL + "resource/api/packages?name=john&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"sw360:packages": [{"name": "john", "version": "2.2.2", "_links": {"self": {"href": "https://my.server.com/resource/api/packages/08ddfd57636c4c47f4c879515007081f"}}}]}}', # noqa status=200, content_type="application/json", @@ -173,7 +173,7 @@ def test_get_packages_by_packagemanager(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/packages?packageManager=nuget", + url=self.MYURL + "resource/api/packages?packageManager=nuget&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"sw360:packages": [{"name": "john", "version": "2.2.2", "_links": {"self": {"href": "https://my.server.com/resource/api/packages/08ddfd57636c4c47f4c879515007081f"}}}]}}', # noqa status=200, content_type="application/json", @@ -195,19 +195,18 @@ def test_get_packages_by_packagemanager_with_details(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/packages?packageManager=nuget&page=1&page_entries=5&sort=name%2Cdesc", + url=self.MYURL + "resource/api/packages?packageManager=nuget&luceneSearch=true&page=1&page_entries=5&sort=name%2Cdesc", body='{"_embedded": {"sw360:packages": [{"name": "john", "version": "2.2.2", "_links": {"self": {"href": "https://my.server.com/resource/api/packages/08ddfd57636c4c47f4c879515007081f"}}}]}}', # noqa status=200, content_type="application/json", adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - packages = lib.get_packages_by_packagemanager("nuget", page=1, page_size=5, sort="name,desc") + packages = lib.get_packages_by_packagemanager("nuget", page=1, page_size=5, sort=PackageSortColumn.NAME.desc()) self.assertIsNotNone(packages) self.assertTrue(len(packages) > 0) - pkgs = packages["_embedded"]["sw360:packages"] - self.assertEqual("john", pkgs[0]["name"]) - self.assertEqual("2.2.2", pkgs[0]["version"]) + self.assertEqual("john", packages[0]["name"]) + self.assertEqual("2.2.2", packages[0]["version"]) @responses.activate def test_create_new_package(self) -> None: diff --git a/tests/test_sw360_projects.py b/tests/test_sw360_projects.py index c1a59f8..7d4e3ca 100644 --- a/tests/test_sw360_projects.py +++ b/tests/test_sw360_projects.py @@ -20,6 +20,7 @@ sys.path.insert(1, "..") from sw360 import SW360, SW360Error # noqa: E402 +from sw360.sorting import ProjectSortColumn # noqa: E402 class Sw360TestProjects(unittest.TestCase): @@ -140,7 +141,7 @@ def test_get_project_releases(self) -> None: lib = self.get_logged_in_lib() responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/123/releases?transitive=false", + url=self.MYURL + "resource/api/projects/123/releases?transitive=false&page=0&page_entries=50&sort=createdOn,desc", body='{"name": "My Testproject"}', status=200, content_type="application/json", @@ -152,7 +153,7 @@ def test_get_project_releases(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/123/releases?transitive=true", + url=self.MYURL + "resource/api/projects/123/releases?transitive=true&page=0&page_entries=50&sort=createdOn,desc", body='{"name": "My Testproject"}', status=200, content_type="application/json", @@ -185,7 +186,7 @@ def test_get_projects(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects", + url=self.MYURL + "resource/api/projects?luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject"}]}}', status=200, content_type="application/json", @@ -195,9 +196,7 @@ def test_get_projects(self) -> None: projects = lib.get_projects() self.assertIsNotNone(projects) if projects: # only for mypy - self.assertTrue("_embedded" in projects) - self.assertTrue("sw360:projects" in projects["_embedded"]) - self.assertEqual("My Testproject", projects["_embedded"]["sw360:projects"][0]["name"]) + self.assertEqual("My Testproject", projects[0]["name"]) @responses.activate def test_get_projects_with_details(self) -> None: @@ -205,7 +204,7 @@ def test_get_projects_with_details(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?allDetails=true", + url=self.MYURL + "resource/api/projects?allDetails=true&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject"}]}}', status=200, content_type="application/json", @@ -215,9 +214,7 @@ def test_get_projects_with_details(self) -> None: projects = lib.get_projects(all_details=True) self.assertIsNotNone(projects) if projects: # only for mypy - self.assertTrue("_embedded" in projects) - self.assertTrue("sw360:projects" in projects["_embedded"]) - self.assertEqual("My Testproject", projects["_embedded"]["sw360:projects"][0]["name"]) + self.assertEqual("My Testproject", projects[0]["name"]) @responses.activate def test_get_projects_with_paging(self) -> None: @@ -225,7 +222,7 @@ def test_get_projects_with_paging(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?page=1&page_entries=2", + url=self.MYURL + "resource/api/projects?luceneSearch=true&page=1&page_entries=2&sort=name,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject"}]}}', status=200, content_type="application/json", @@ -235,9 +232,7 @@ def test_get_projects_with_paging(self) -> None: projects = lib.get_projects(page=1, page_size=2) self.assertIsNotNone(projects) if projects: # only for mypy - self.assertTrue("_embedded" in projects) - self.assertTrue("sw360:projects" in projects["_embedded"]) - self.assertEqual("My Testproject", projects["_embedded"]["sw360:projects"][0]["name"]) + self.assertEqual("My Testproject", projects[0]["name"]) @responses.activate def test_get_projects_with_paging_and_details(self) -> None: @@ -245,19 +240,17 @@ def test_get_projects_with_paging_and_details(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?allDetails=true&page=3&page_entries=4&sort=name%2Cdesc", # noqa + url=self.MYURL + "resource/api/projects?allDetails=true&luceneSearch=true&page=3&page_entries=4&sort=name%2Cdesc", # noqa body='{"_embedded": {"sw360:projects": [{"name": "My Testproject"}]}}', status=200, content_type="application/json", adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - projects = lib.get_projects(all_details=True, page=3, page_size=4, sort="name,desc") + projects = lib.get_projects(all_details=True, page=3, page_size=4, sort=ProjectSortColumn.NAME.desc()) self.assertIsNotNone(projects) if projects: # only for mypy - self.assertTrue("_embedded" in projects) - self.assertTrue("sw360:projects" in projects["_embedded"]) - self.assertEqual("My Testproject", projects["_embedded"]["sw360:projects"][0]["name"]) + self.assertEqual("My Testproject", projects[0]["name"]) @responses.activate def test_get_projects_by_type(self) -> None: @@ -265,7 +258,7 @@ def test_get_projects_by_type(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?type=SERVICE", + url=self.MYURL + "resource/api/projects?type=SERVICE&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject", "projectType": "SERVICE"}]}}', # noqa status=200, content_type="application/json", @@ -284,7 +277,7 @@ def test_get_projects_by_type_no_reply(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?type=SERVICE", + url=self.MYURL + "resource/api/projects?type=SERVICE&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{}', status=200, content_type="application/json", @@ -301,7 +294,7 @@ def test_get_projects_by_type_invalid_reply(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?type=SERVICE", + url=self.MYURL + "resource/api/projects?type=SERVICE&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_xxembedded": {"sw360:projects": [{"name": "My Testproject", "projectType": "SERVICE"}]}}', # noqa status=200, content_type="application/json", @@ -318,7 +311,7 @@ def test_get_projects_by_type_invalid_reply2(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?type=SERVICE", + url=self.MYURL + "resource/api/projects?type=SERVICE&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"xxsw360:projects": [{"name": "My Testproject", "projectType": "SERVICE"}]}}', # noqa status=200, content_type="application/json", @@ -335,7 +328,7 @@ def test_get_project_names(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects", + url=self.MYURL + "resource/api/projects?luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject", "version" : "1.0.0"}]}}', # noqa status=200, content_type="application/json", @@ -404,7 +397,7 @@ def test_get_projects_by_name(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?name=My", + url=self.MYURL + "resource/api/projects?name=My&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{}', status=200, content_type="application/json", @@ -420,7 +413,7 @@ def test_get_projects_by_name_no_result(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?name=My", + url=self.MYURL + "resource/api/projects?name=My&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject"}]}}', status=200, content_type="application/json", @@ -438,7 +431,7 @@ def test_get_projects_by_name_invalid_reply(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?name=My", + url=self.MYURL + "resource/api/projects?name=My&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_xxembedded": {"sw360:projects": [{"name": "My Testproject"}]}}', status=200, content_type="application/json", @@ -455,7 +448,7 @@ def test_get_projects_by_name_invalid_reply2(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?name=My", + url=self.MYURL + "resource/api/projects?name=My&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"xxsw360:projects": [{"name": "My Testproject"}]}}', status=200, content_type="application/json", @@ -540,7 +533,7 @@ def test_get_projects_by_group(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?group=SI", + url=self.MYURL + "resource/api/projects?group=SI&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject", "externalIds": {"com.siemens.code.project.id": "13171"}}]}}', # noqa status=200, content_type="application/json", @@ -558,7 +551,7 @@ def test_get_projects_by_group_invalid_reply(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?group=SI", + url=self.MYURL + "resource/api/projects?group=SI&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_xxembedded": {"sw360:projects": [{"name": "My Testproject", "externalIds": {"com.siemens.code.project.id": "13171"}}]}}', # noqa status=200, content_type="application/json", @@ -575,7 +568,7 @@ def test_get_projects_by_group_invalid_reply2(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?group=SI", + url=self.MYURL + "resource/api/projects?group=SI&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"xxsw360:projects": [{"name": "My Testproject", "externalIds": {"com.siemens.code.project.id": "13171"}}]}}', # noqa status=200, content_type="application/json", @@ -592,7 +585,7 @@ def test_get_projects_by_group_with_details(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?allDetails?group=SI", + url=self.MYURL + "resource/api/projects?allDetails=true&group=SI&luceneSearch=true&page=0&page_entries=50&sort=score,asc", # noqa: E501 body='{"_embedded": {"sw360:projects": [{"name": "My Testproject", "externalIds": {"com.siemens.code.project.id": "13171"}}]}}', # noqa status=200, content_type="application/json", @@ -610,7 +603,7 @@ def test_get_projects_by_group_no_result(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?group=SI", + url=self.MYURL + "resource/api/projects?group=SI&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{}', status=200, content_type="application/json", @@ -626,7 +619,7 @@ def test_get_projects_by_tag(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?tag=SI BP&luceneSearch=true", + url=self.MYURL + "resource/api/projects?tag=SI BP&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject", "externalIds": {"com.siemens.code.project.id": "13171"}}]}}', # noqa status=200, content_type="application/json", @@ -644,7 +637,7 @@ def test_get_projects_by_tag_no_result(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?tag=SI&luceneSearch=true", + url=self.MYURL + "resource/api/projects?tag=SI&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{}', status=200, content_type="application/json", @@ -660,7 +653,7 @@ def test_get_projects_by_tag_invalid_reply(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?tag=SI BP&luceneSearch=true", + url=self.MYURL + "resource/api/projects?tag=SI BP&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_xxembedded": {"sw360:projects": [{"name": "My Testproject", "externalIds": {"com.siemens.code.project.id": "13171"}}]}}', # noqa status=200, content_type="application/json", @@ -677,7 +670,7 @@ def test_get_projects_by_tag_invalid_reply2(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?tag=SI BP&luceneSearch=true", + url=self.MYURL + "resource/api/projects?tag=SI BP&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"xxsw360:projects": [{"name": "My Testproject", "externalIds": {"com.siemens.code.project.id": "13171"}}]}}', # noqa status=200, content_type="application/json", @@ -718,7 +711,7 @@ def test_get_project_vulnerabilities(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/123/vulnerabilities", + url=self.MYURL + "resource/api/projects/123/vulnerabilities?page=0&page_entries=50", body='{"_embedded": {"sw360:vulnerabilityDToes": [{"priority": "2 - major", "action": "Follow Recommendation"}]}}', # noqa status=200, content_type="application/json", @@ -741,7 +734,7 @@ def test_get_project_vulnerabilities_no_result(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/123/vulnerabilities", + url=self.MYURL + "resource/api/projects/123/vulnerabilities?&page=0&page_entries=50", body='{}', status=200, content_type="application/json", @@ -1011,7 +1004,7 @@ def test_update_project_releases_fresh_prj(self) -> None: lib = self.get_logged_in_lib() responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/123/releases?transitive=false", + url=self.MYURL + "resource/api/projects/123/releases?transitive=false&page=0&page_entries=50&sort=createdOn,desc", json={}, ) responses.add( @@ -1023,7 +1016,7 @@ def test_update_project_releases_fresh_prj(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/124/releases?transitive=false", + url=self.MYURL + "resource/api/projects/124/releases?transitive=false&page=0&page_entries=50&sort=createdOn,desc", json={'_embedded': {'sw360:projects': []}}, ) responses.add( @@ -1039,7 +1032,7 @@ def test_update_project_releases(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/123/releases?transitive=false", + url=self.MYURL + "resource/api/projects/123/releases?transitive=false&page=0&page_entries=50&sort=createdOn,desc", body='{"_embedded": {"sw360:releases": [{"name": "ngx-device-detector ", "version": "1.3.20","_links": {"self": {"href": "https://sw360.siemens.com/resource/api/releases/3a4865e453873ee00d924469ff40f391" }}}]}}', # noqa status=200, content_type="application/json", @@ -1118,7 +1111,7 @@ def test_update_project_releases_failed(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/123/releases?transitive=false", + url=self.MYURL + "resource/api/projects/123/releases?transitive=false&page=0&page_entries=50&sort=createdOn,desc", body='{"_embedded": {"sw360:releases": [{"name": "ngx-device-detector ", "version": "1.3.20","_links": {"self": {"href": "https://sw360.siemens.com/resource/api/releases/3a4865e453873ee00d924469ff40f391" }}}]}}', # noqa status=200, content_type="application/json", diff --git a/tests/test_sw360_releases.py b/tests/test_sw360_releases.py index d4191e5..d6974de 100644 --- a/tests/test_sw360_releases.py +++ b/tests/test_sw360_releases.py @@ -17,6 +17,7 @@ sys.path.insert(1, "..") from sw360 import SW360, SW360Error # noqa: E402 +from sw360.sorting import ReleaseSortColumn # noqa: E402 class Sw360TestReleases(unittest.TestCase): @@ -161,7 +162,7 @@ def test_get_all_releases_isnewclearing_with_source_available(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/releases?isNewClearingWithSourceAvailable=true", + url=self.MYURL + "resource/api/releases?isNewClearingWithSourceAvailable=true&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:releases": [{"name": "Tethys.Logging", "version": "1.3.0"}]}}', status=200, content_type="application/json", @@ -185,7 +186,7 @@ def test_get_all_releases_all_details(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/releases?allDetails=true", + url=self.MYURL + "resource/api/releases?allDetails=true&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:releases": [{"name": "Tethys.Logging", "version": "1.3.0", "releaseDate": "2018-03-04"}]}}', # noqa status=200, content_type="application/json", @@ -207,7 +208,7 @@ def test_get_all_releases_with_fields(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/releases?fields=releaseDate", + url=self.MYURL + "resource/api/releases?fields=releaseDate&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:releases": [{"name": "Tethys.Logging", "version": "1.3.0", "releaseDate": "2018-03-04"}]}}', # noqa status=200, content_type="application/json", @@ -229,7 +230,7 @@ def test_get_all_releases_with_fields_and_all_details(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/releases?allDetails=true&fields=releaseDate", + url=self.MYURL + "resource/api/releases?allDetails=true&fields=releaseDate&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:releases": [{"name": "Tethys.Logging", "version": "1.3.0", "releaseDate": "2018-03-04"}]}}', # noqa status=200, content_type="application/json", @@ -251,19 +252,18 @@ def test_get_all_releases_with_paging_and_sorting(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/releases?page=2&page_entries=5&sort=name%2Casc", + url=self.MYURL + "resource/api/releases?page=2&page_entries=5&sort=name%2Casc&luceneSearch=true", body='{"_embedded": {"sw360:releases": [{"name": "Tethys.Logging", "version": "1.3.0", "releaseDate": "2018-03-04"}]}}', # noqa status=200, content_type="application/json", adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - releases = lib.get_all_releases(page=2, page_size=5, sort="name,asc") + releases = lib.get_all_releases(page=2, page_size=5, sort=ReleaseSortColumn.NAME.asc()) self.assertIsNotNone(releases) self.assertTrue(len(releases) > 0) - rel = releases["_embedded"]["sw360:releases"] - self.assertEqual("Tethys.Logging", rel[0]["name"]) - self.assertEqual("1.3.0", rel[0]["version"]) + self.assertEqual("Tethys.Logging", releases[0]["name"]) + self.assertEqual("1.3.0", releases[0]["version"]) @responses.activate def test_get_releases_by_external_id(self) -> None: @@ -316,7 +316,7 @@ def test_get_releases_by_name(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/releases?name=john", + url=self.MYURL + "resource/api/releases?name=john&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"sw360:releases": [{"name": "john", "version": "2.2.2", "_links": {"self": {"href": "https://my.server.com/resource/api/releases/08ddfd57636c4c47f4c879515007081f"}}}]}}', # noqa status=200, content_type="application/json", @@ -338,7 +338,7 @@ def test_get_releases_by_name_invalid_answer(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/releases?name=john", + url=self.MYURL + "resource/api/releases?name=john&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_xxembedded": {"sw360:releases": [{"name": "john", "version": "2.2.2", "_links": {"self": {"href": "https://my.server.com/resource/api/releases/08ddfd57636c4c47f4c879515007081f"}}}]}}', # noqa status=200, content_type="application/json",