feat(vuln): update API calls to v20 style - #63
Open
GMishx wants to merge 9 commits into
Open
Conversation
1. Update the functions exposed for Projects to use v20 style API calls. 2. Add pagination to calls where supported. 3. Add new utility for uploading attachments. Signed-off-by: Gaurav Mishra <mishra.gaurav@siemens.com>
1. Update the functions exposed for Components to use v20 style API calls. 2. Add pagination to calls where supported. 3. Add new utility for uploading attachments. Signed-off-by: Gaurav Mishra <mishra.gaurav@siemens.com> diff --git c/sw360/components.py i/sw360/components.py index 6792d6e..59d945e 100644 --- c/sw360/components.py +++ i/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(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 @@ class ComponentsMixin(BaseMixin): 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 + if sort is None: + sort = ComponentSortColumn.NAME.asc() - 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"] - - 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 @@ class ComponentsMixin(BaseMixin): fullbase_url = self.url + "resource/api/components" params = {"type": component_type} - 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.NAME.asc() - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) - - 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 @@ class ComponentsMixin(BaseMixin): 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 @@ class ComponentsMixin(BaseMixin): 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 @@ class ComponentsMixin(BaseMixin): 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 @@ class ComponentsMixin(BaseMixin): 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 @@ class ComponentsMixin(BaseMixin): :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 @@ class ComponentsMixin(BaseMixin): 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 c/sw360/sorting.py i/sw360/sorting.py index 0494661..224b821 100644 --- c/sw360/sorting.py +++ i/sw360/sorting.py @@ -44,3 +44,11 @@ class ReleaseSortColumn(BaseSortMixin, Enum): 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" diff --git c/tests/test_sw360_components.py i/tests/test_sw360_components.py index 8d0d209..b9fde6c 100644 --- c/tests/test_sw360_components.py +++ i/tests/test_sw360_components.py @@ -18,6 +18,7 @@ import responses 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 @@ class Sw360TestComponents(unittest.TestCase): 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 @@ class Sw360TestComponents(unittest.TestCase): 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&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 @@ class Sw360TestComponents(unittest.TestCase): 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 @@ class Sw360TestComponents(unittest.TestCase): 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 @@ class Sw360TestComponents(unittest.TestCase): 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 @@ class Sw360TestComponents(unittest.TestCase): 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 @@ class Sw360TestComponents(unittest.TestCase): 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 @@ class Sw360TestComponents(unittest.TestCase): 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 @@ class Sw360TestComponents(unittest.TestCase): 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 @@ class Sw360TestComponents(unittest.TestCase): 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 @@ class Sw360TestComponents(unittest.TestCase): 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",
Signed-off-by: Gaurav Mishra <mishra.gaurav@siemens.com>
Signed-off-by: Gaurav Mishra <mishra.gaurav@siemens.com>
Signed-off-by: Gaurav Mishra <mishra.gaurav@siemens.com>
Signed-off-by: Gaurav Mishra <mishra.gaurav@siemens.com>
Signed-off-by: Gaurav Mishra <mishra.gaurav@siemens.com>
Signed-off-by: Gaurav Mishra <mishra.gaurav@siemens.com>
Signed-off-by: Gaurav Mishra <mishra.gaurav@siemens.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Add API v20 specific changes for Vulnerability module.
This PR builds on top of #62