Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

add recentreleases endpoint #36

Merged
merged 6 commits into from
Oct 26, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion ChangeLog.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,15 @@

# SW360 Base Library for Python

## NEXT

* more REST API endpoints implemented:
* `get_recent_releases`
* `get_recent_components`
* `get_all_moderation_requests`
* `get_moderation_requests_by_state`
* `get_moderation_request`

## V1.6.0

* packages REST API calls implemented.
Expand All @@ -21,7 +30,7 @@

* when using CaPyCLI in a CI pipeline, connection problems to the SW360 server (5xx) cause
the pipeline to fail. We have now add an improved session handling to all api requests.
* dependency updates due to security vulnerabilities in idna.
* dependency updates due to security vulnerabilities in `idna`.

## V1.4.1

Expand Down
12 changes: 3 additions & 9 deletions sw360/attachments.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,16 +55,10 @@ def get_attachment_infos_for_resource(self, resource_type: str, resource_id: str
+ "/attachments"
)

if not resp:
return []
if resp and "_embedded" in resp and "sw360:attachments" in resp["_embedded"]:
return resp["_embedded"]["sw360:attachments"]

if "_embedded" not in resp:
return []

if "sw360:attachments" not in resp["_embedded"]:
return []

return resp["_embedded"]["sw360:attachments"]
return []

def get_attachment_infos_for_release(self, release_id: str) -> List[Dict[str, Any]]:
"""Get information about the attachments of a release
Expand Down
30 changes: 20 additions & 10 deletions sw360/components.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
# -------------------------------------------------------------------------------
# Copyright (c) 2019-2023 Siemens
# Copyright (c) 2019-2024 Siemens
# Copyright (c) 2022 BMW CarIT GmbH
# All Rights Reserved.
# Authors: [email protected], [email protected]
Expand Down Expand Up @@ -84,16 +84,10 @@ def get_components_by_type(self, component_type: str) -> List[Dict[str, Any]]:
"""

resp = self.api_get(self.url + "resource/api/components?type=" + component_type)
if not resp:
return []

if "_embedded" not in resp:
return []

if "sw360:components" not in resp["_embedded"]:
return []
if resp and ("_embedded" in resp) and ("sw360:components" in resp["_embedded"]):
return resp["_embedded"]["sw360:components"]

return resp["_embedded"]["sw360:components"]
return []

def get_component(self, component_id: str) -> Optional[Dict[str, Any]]:
"""Get information of about a component
Expand Down Expand Up @@ -292,3 +286,19 @@ def get_users_of_component(self, component_id: str) -> Optional[Dict[str, Any]]:

resp = self.api_get(self.url + "resource/api/components/usedBy/" + component_id)
return resp

def get_recent_components(self) -> Optional[Dict[str, Any]]:
"""Get 5 of the service's most recently created components.

API endpoint: GET /components/recentComponents

:return: a list of components
:rtype: JSON list of component objects
:raises SW360Error: if there is a negative HTTP response
"""
url = self.url + "resource/api/components/recentComponents"
resp = self.api_get(url)
if resp and ("_embedded" in resp) and ("sw360:components" in resp["_embedded"]):
return resp["_embedded"]["sw360:components"]

return resp
12 changes: 3 additions & 9 deletions sw360/license.py
Original file line number Diff line number Diff line change
Expand Up @@ -117,16 +117,10 @@ def get_all_licenses(self) -> List[Dict[str, Any]]:
"""

resp = self.api_get(self.url + "resource/api/licenses")
if not resp:
return []
if resp and "_embedded" in resp and "sw360:licenses" in resp["_embedded"]:
return resp["_embedded"]["sw360:licenses"]

if "_embedded" not in resp:
return []

if "sw360:licenses" not in resp["_embedded"]:
return []

return resp["_embedded"]["sw360:licenses"]
return []

def get_license(self, license_id: str) -> Optional[Dict[str, Any]]:
"""Get information of about a license
Expand Down
95 changes: 95 additions & 0 deletions sw360/moderationrequests.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,95 @@
# -------------------------------------------------------------------------------
# Copyright (c) 2024 Siemens
# All Rights Reserved.
# Authors: [email protected]
#
# Licensed under the MIT license.
# SPDX-License-Identifier: MIT
# -------------------------------------------------------------------------------

from typing import Any, Dict, List, Optional

from .base import BaseMixin


class ModerationRequestMixin(BaseMixin):
def get_all_moderation_requests(self, page: int = -1, page_size: int = -1,
sort: str = "") -> List[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
:type page_size: int
:param sort: sort order for the packages ("name,desc"; "name,asc")
:type sort: str
:return: list of moderation requests
:rtype: list of JSON moderation requests objects
:raises SW360Error: if there is a negative HTTP response
"""

full_url = self.url + "resource/api/moderationrequest"
if page > -1:
full_url = self._add_param(full_url, "page=" + str(page))
full_url = self._add_param(full_url, "page_entries=" + str(page_size))

if sort:
# ensure HTML encoding
sort = sort.replace(",", "%2C")
full_url = self._add_param(full_url, "sort=" + sort)

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 = "") -> List[Dict[str, Any]]:
"""Get information of about all moderation requests

API endpoint: GET /moderationrequest/byState

: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
:type page_size: int
:param sort: sort order for the packages ("name,desc"; "name,asc")
:type sort: str
:return: list of moderation requests
:rtype: list of JSON moderation requests objects
:raises SW360Error: if there is a negative HTTP response
"""

full_url = self.url + "resource/api/moderationrequest/byState?state=" + state
if all_details:
full_url = self._add_param(full_url, "allDetails=true")

if page > -1:
full_url = self._add_param(full_url, "page=" + str(page))
full_url = self._add_param(full_url, "page_entries=" + str(page_size))

if sort:
# ensure HTML encoding
sort = sort.replace(",", "%2C")
full_url = self._add_param(full_url, "sort=" + sort)

resp = self.api_get(full_url)
return resp

def get_moderation_request(self, mr_id: str) -> Optional[Dict[str, Any]]:
"""Get information of about a moderation request

API endpoint: GET /moderationrequest/{id}

:param mr_id: the id of the moderation request to be requested
:type mr_id: string
:return: a moderation request
:rtype: JSON moderation request object
:raises SW360Error: if there is a negative HTTP response
"""

resp = self.api_get(self.url + "resource/api/moderationrequest/" + mr_id)
return resp
16 changes: 16 additions & 0 deletions sw360/releases.py
Original file line number Diff line number Diff line change
Expand Up @@ -286,3 +286,19 @@ def unlink_packages_from_release(self, release_id: str, packages: List[str]) ->

url = self.url + "resource/api/releases/" + release_id + "/unlink/packages/"
return self.api_patch(url, json=packages)

def get_recent_releases(self) -> Optional[Dict[str, Any]]:
"""Get 5 of the service's most recently created releases.

API endpoint: GET /releases/recentReleases

:return: a list of releases
:rtype: JSON list of release objects
:raises SW360Error: if there is a negative HTTP response
"""
url = self.url + "resource/api/releases/recentReleases"
resp = self.api_get(url)
if resp and ("_embedded" in resp) and ("sw360:releases" in resp["_embedded"]):
return resp["_embedded"]["sw360:releases"]

return resp
4 changes: 3 additions & 1 deletion sw360/sw360_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
from .sw360error import SW360Error
from .vendor import VendorMixin
from .vulnerabilities import VulnerabilitiesMixin
from .moderationrequests import ModerationRequestMixin

# Retry mechanism for rate limiting
adapter = HTTPAdapter(max_retries=Retry(
Expand All @@ -48,7 +49,8 @@ class SW360(
ReleasesMixin,
VendorMixin,
VulnerabilitiesMixin,
PackagesMixin
PackagesMixin,
ModerationRequestMixin
):
"""Python interface to the Siemens SW360 platform

Expand Down
12 changes: 3 additions & 9 deletions sw360/vendor.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,16 +27,10 @@ def get_all_vendors(self) -> List[Dict[str, Any]]:
"""

resp = self.api_get(self.url + "resource/api/vendors")
if not resp:
return []
if resp and "_embedded" in resp and "sw360:vendors" in resp["_embedded"]:
return resp["_embedded"]["sw360:vendors"]

if "_embedded" not in resp:
return []

if "sw360:vendors" not in resp["_embedded"]:
return []

return resp["_embedded"]["sw360:vendors"]
return []

def get_vendor(self, vendor_id: str) -> Optional[Dict[str, Any]]:
"""Returns a vendor
Expand Down
66 changes: 64 additions & 2 deletions tests/test_sw360_components.py
Original file line number Diff line number Diff line change
Expand Up @@ -839,11 +839,73 @@ def xtest_get_component_real(self) -> None:
lib.login_api()
# c = lib.get_component("eaba2f0416e000e8ca5b2ccb4400633e")
# c = lib.get_components_by_external_id( "package-url", "pkg:nuget/Tethys.Logging")
c = lib.api_get("https://sw360.siemens.com/resource/api/components/searchByExternalIds?package-url=pkg:nuget/Tethys.Logging") # noqa
c = lib.api_get("https://my.server.com/resource/api/components/searchByExternalIds?package-url=pkg:nuget/Tethys.Logging") # noqa
print(c)

@responses.activate
def test_get_recent_components(self) -> None:
lib = SW360(self.MYURL, self.MYTOKEN, False)
lib.force_no_session = True
self._add_login_response()
actual = lib.login_api()
self.assertTrue(actual)

responses.add(
method=responses.GET,
url=self.MYURL + "resource/api/components/recentComponents",
body='''{
"_embedded": {
"sw360:components": [
{
"id": "ff6f1b5b212b4f93b306e2cceca4f64d",
"name": "intl-listformat",
"description": "n/a",
"componentType": "OSS",
"visbility": "EVERYONE",
"mainLicenseIds": [],
"_links": {
"self": {
"href": "https://my.server.com/resource/api/components/ff"
}
}
},
{
"id": "f916b35d6c864014bd8823c45615aeab",
"name": "fields-metadata-plugin",
"description": "n/a",
"componentType": "OSS",
"visbility": "EVERYONE",
"mainLicenseIds": [],
"_links": {
"self": {
"href": "https://my.server.com/resource/api/components/f9"
}
}
}
]
},
"_links": {
"curies": [
{
"href": "https://my.server.com/resource/docs/{rel}.html",
"name": "sw360",
"templated": true
}
]
}
}''',
status=200,
content_type="application/json",
adding_headers={"Authorization": "Token " + self.MYTOKEN},
)

components = lib.get_recent_components()
self.assertIsNotNone(components)
self.assertEqual(2, len(components))
self.assertEqual("intl-listformat", components[0]["name"])
self.assertEqual("OSS", components[0]["componentType"])


if __name__ == "__main__":
# unittest.main()
x = Sw360TestComponents()
x.test_get_all_components_with_fields_and_paging()
Loading
Loading