From a56613fecb5ef27dabd2f99b5a0d757346554f04 Mon Sep 17 00:00:00 2001 From: Marion Deveaud Date: Sun, 12 Nov 2023 11:30:08 +0100 Subject: [PATCH] feat(items): prepare new item endpoints for the next release --- fossology/__init__.py | 5 +- fossology/enums.py | 11 + fossology/items.py | 254 ++++++++++++++++++ fossology/obj.py | 44 +++ fossology/uploads.py | 172 +----------- ...{test_upload_clearing.py => test_items.py} | 70 ++++- 6 files changed, 383 insertions(+), 173 deletions(-) create mode 100644 fossology/items.py rename tests/{test_upload_clearing.py => test_items.py} (74%) diff --git a/fossology/__init__.py b/fossology/__init__.py index a1567fa..ce9f540 100644 --- a/fossology/__init__.py +++ b/fossology/__init__.py @@ -10,6 +10,7 @@ from fossology.exceptions import AuthenticationError, FossologyApiError from fossology.folders import Folders from fossology.groups import Groups +from fossology.items import Items from fossology.jobs import Jobs from fossology.license import LicenseEndpoint from fossology.obj import Agents, ApiInfo, HealthInfo, User @@ -79,7 +80,9 @@ def fossology_token( exit(f"Server {url} does not seem to be running or is unreachable: {error}") -class Fossology(Folders, Groups, LicenseEndpoint, Uploads, Jobs, Report, Users, Search): +class Fossology( + Folders, Groups, Items, LicenseEndpoint, Uploads, Jobs, Report, Users, Search +): """Main Fossology API class diff --git a/fossology/enums.py b/fossology/enums.py index 19a0e47..d33e974 100644 --- a/fossology/enums.py +++ b/fossology/enums.py @@ -195,3 +195,14 @@ class PrevNextSelection(Enum): WITHLICENSES = "withLicenses" NOCLEARING = "noClearing" + + +class CopyrightStatus(Enum): + """Status of the copyrights: + + ACTIVE + INACTIVE + """ + + ACTIVE = "active" + INACTIVE = "inactive" diff --git a/fossology/items.py b/fossology/items.py new file mode 100644 index 0000000..26e0063 --- /dev/null +++ b/fossology/items.py @@ -0,0 +1,254 @@ +# mypy: disable-error-code="attr-defined" +# Copyright 2019 Siemens AG +# SPDX-License-Identifier: MIT +import json +import logging + +from fossology.enums import CopyrightStatus, PrevNextSelection +from fossology.exceptions import FossologyApiError +from fossology.obj import ( + FileInfo, + GetBulkHistory, + GetClearingHistory, + GetPrevNextItem, + Upload, +) + +logger = logging.getLogger(__name__) +logger.setLevel(logging.DEBUG) + + +class Items: + """Class dedicated to all "uploads...items" related endpoints""" + + def item_info( + self, + upload: Upload, + item_id: int, + ) -> FileInfo: + """Get the info for a specific upload item + + API Endpoint: GET /uploads/{id}/item/{itemId}/info + + :param upload: the upload to get items from + :param item_id: the id of the item + :type upload: Upload + :type item_id: int, + :return: the file info for the specified item + :rtype: FileInfo + :raises FossologyApiError: if the REST call failed + """ + response = self.session.get( + f"{self.api}/uploads/{upload.id}/item/{item_id}/info" + ) + + if response.status_code == 200: + return FileInfo.from_json(response.json()) + + elif response.status_code == 404: + description = f"Upload {upload.id} or item {item_id} not found" + raise FossologyApiError(description, response) + else: + print(response.json()) + description = f"API error while getting info for item {item_id} from upload {upload.uploadname}." + raise FossologyApiError(description, response) + + def item_copyrights( + self, + upload: Upload, + item_id: int, + status: CopyrightStatus, + ) -> int: + """Get the total copyrights of the mentioned upload tree ID + + API Endpoint: GET /uploads/{id}/item/{itemId}/totalcopyrights + + :param upload: the upload to get items from + :param item_id: the id of the item + :param status: the status of the copyrights + :type upload: Upload + :type item_id: int, + :return: the total number of copyrights for the uploadtree item + :rtype: int + :raises FossologyApiError: if the REST call failed + """ + response = self.session.get( + f"{self.api}/uploads/{upload.id}/item/{item_id}/totalcopyrights?status={status.value}" + ) + + if response.status_code == 200: + return response.json()["total_copyrights"] + + elif response.status_code == 404: + description = f"Upload {upload.id} or item {item_id} not found" + raise FossologyApiError(description, response) + else: + description = f"API error while getting total copyrights for item {item_id} from upload {upload.uploadname}." + raise FossologyApiError(description, response) + + def get_clearing_history( + self, + upload: Upload, + item_id: int, + ) -> list[GetClearingHistory]: + """Get the clearing history for a specific upload item + + API Endpoint: GET /uploads/{id}/item/{itemId}/clearing-history + + :param upload: the upload to get items from + :param item_id: the id of the item with clearing decision + :type upload: Upload + :type item_id: int, + :return: the clearing history for the specified item + :rtype: List[GetClearingHistory] + :raises FossologyApiError: if the REST call failed + :raises AuthorizationError: if the REST call is not authorized + """ + response = self.session.get( + f"{self.api}/uploads/{upload.id}/item/{item_id}/clearing-history" + ) + + if response.status_code == 200: + clearing_history = [] + for action in response.json(): + clearing_history.append(GetClearingHistory.from_json(action)) + return clearing_history + + elif response.status_code == 404: + description = f"Upload {upload.id} or item {item_id} not found" + raise FossologyApiError(description, response) + else: + description = f"API error while getting clearing history for item {item_id} from upload {upload.uploadname}." + raise FossologyApiError(description, response) + + def get_prev_next( + self, upload: Upload, item_id: int, selection: PrevNextSelection | None = None + ) -> GetPrevNextItem: + """Get the index of the previous and the next time for an upload + + API Endpoint: GET /uploads/{id}/item/{itemId}/prev-next + + :param upload: the upload to get items from + :param item_id: the id of the item with clearing decision + :param selection: tell Fossology server how to select prev-next item + :type upload: Upload + :type item_id: int + :type selection: str + :return: list of items for the clearing history + :rtype: List[GetPrevNextItem] + :raises FossologyApiError: if the REST call failed + :raises AuthorizationError: if the REST call is not authorized + """ + params = {} + if selection: + params["selection"] = selection + + response = self.session.get( + f"{self.api}/uploads/{upload.id}/item/{item_id}/prev-next", params=params + ) + + if response.status_code == 200: + return GetPrevNextItem.from_json(response.json()) + + elif response.status_code == 404: + description = f"Upload {upload.id} or item {item_id} not found" + raise FossologyApiError(description, response) + else: + description = f"API error while getting prev-next items for {item_id} from upload {upload.uploadname}." + raise FossologyApiError(description, response) + + def get_bulk_history( + self, + upload: Upload, + item_id: int, + ) -> list[GetBulkHistory]: + """Get the bulk history for a specific upload item + + API Endpoint: GET /uploads/{id}/item/{itemId}/bulk-history + + :param upload: the upload to get items from + :param item_id: the id of the item with clearing decision + :type upload: Upload + :type item_id: int + :return: list of data from the bulk history + :rtype: List[GetBulkHistory] + :raises FossologyApiError: if the REST call failed + :raises AuthorizationError: if the REST call is not authorized + """ + response = self.session.get( + f"{self.api}/uploads/{upload.id}/item/{item_id}/bulk-history" + ) + + if response.status_code == 200: + bulk_history = [] + for item in response.json(): + bulk_history.append(GetBulkHistory.from_json(item)) + return bulk_history + + elif response.status_code == 404: + description = f"Upload {upload.id} or item {item_id} not found" + raise FossologyApiError(description, response) + else: + description = f"API error while getting bulk history for {item_id} from upload {upload.uploadname}." + raise FossologyApiError(description, response) + + def schedule_bulk_scan( + self, + upload: Upload, + item_id: int, + spec: dict, + ): + """Schedule a bulk scan for a specific upload item + + API Endpoint: POST /uploads/{id}/item/{itemId}/bulk-scan + + Bulk scan specifications `spec` are added to the request body, + following options are available: + + >>> bulk_scan_spec = { + ... "bulkActions": [ + ... { + ... "licenseShortName": 'MIT', + ... "licenseText": 'License text', + ... "acknowledgement": 'Acknowledgment text', + ... "comment": 'Comment text', + ... "licenseAction": 'ADD', # or 'REMOVE' + ... } + ... ], + ... "refText": 'Reference Text', + ... "bulkScope": 'folder', # or upload + ... "forceDecision": 'false', + ... "ignoreIrre": 'false', + ... "delimiters": 'DEFAULT', + ... "scanOnlyFindings": 'true', + ... } + + :param upload: the upload for the bulk scan + :param item_id: the id of the item for the bulk scan + :param spec: bulk scan specification + :type upload: Upload + :type item_id: int + :raises FossologyApiError: if the REST call failed + :raises AuthorizationError: if the REST call is not authorized + """ + headers = {"Content-Type": "application/json"} + response = self.session.post( + f"{self.api}/uploads/{upload.id}/item/{item_id}/bulk-scan", + headers=headers, + data=json.dumps(spec), + ) + if response.status_code == 201: + logger.info( + f"Bulk scan scheduled for upload {upload.uploadname}, item {item_id}" + ) + elif response.status_code == 400: + description = ( + f"Bad bulk scan request for upload {upload.id}, item {item_id}" + ) + raise FossologyApiError(description, response) + elif response.status_code == 404: + description = f"Upload {upload.id} or item {item_id} not found" + raise FossologyApiError(description, response) + else: + description = f"API error while scheduling bulk scan for item {item_id} from upload {upload.uploadname}." + raise FossologyApiError(description, response) diff --git a/fossology/obj.py b/fossology/obj.py index 44c808f..ae40462 100644 --- a/fossology/obj.py +++ b/fossology/obj.py @@ -530,6 +530,50 @@ def from_json(cls, json_dict): return cls(**json_dict) +class FileInfo(object): + + """FOSSology file info response. + + Represents a FOSSology file info response. + + :param view_info: view info of the file + :param meta_info: meta info of the file + :param package_info: package info of the file + :param tag_info: tag info of the file + :param reuse_info: reuse info of the file + :param kwargs: handle any other license information provided by the fossology instance + :type view_info: Object + :type meta_info: Object + :type package_info: Object + :type tag_info: Object + :type reuse_info: Object + :type kwargs: key word argument + """ + + def __init__( + self, + view_info, + meta_info, + package_info, + tag_info, + reuse_info, + **kwargs, + ): + self.view_info = view_info + self.meta_info = meta_info + self.package_info = package_info + self.tag_info = tag_info + self.reuse_info = reuse_info + self.additional_info = kwargs + + def __str__(self): + return f"File view {self.view_info}" + + @classmethod + def from_json(cls, json_dict): + return cls(**json_dict) + + class Upload(object): """FOSSology upload. diff --git a/fossology/uploads.py b/fossology/uploads.py index de96ae2..98ba3fe 100644 --- a/fossology/uploads.py +++ b/fossology/uploads.py @@ -10,13 +10,10 @@ import requests from tenacity import TryAgain, retry, retry_if_exception_type, stop_after_attempt -from fossology.enums import AccessLevel, ClearingStatus, PrevNextSelection +from fossology.enums import AccessLevel, ClearingStatus from fossology.exceptions import AuthorizationError, FossologyApiError from fossology.obj import ( Folder, - GetBulkHistory, - GetClearingHistory, - GetPrevNextItem, Group, Permission, Summary, @@ -792,170 +789,3 @@ def upload_permissions( f"API error while getting permissions for upload {upload.uploadname}." ) raise FossologyApiError(description, response) - - def get_clearing_history( - self, - upload: Upload, - item_id: int, - ) -> list[GetClearingHistory]: - """Get the clearing history for a specific upload item - - API Endpoint: GET /uploads/{id}/item/{itemId}/clearing-history - - :param upload: the upload to get items from - :param item_id: the id of the item with clearing decision - :type upload: Upload - :type item_id: int, - :return: the clearing history for the specified item - :rtype: List[GetClearingHistory] - :raises FossologyApiError: if the REST call failed - :raises AuthorizationError: if the REST call is not authorized - """ - response = self.session.get( - f"{self.api}/uploads/{upload.id}/item/{item_id}/clearing-history" - ) - - if response.status_code == 200: - clearing_history = [] - for action in response.json(): - clearing_history.append(GetClearingHistory.from_json(action)) - return clearing_history - - elif response.status_code == 404: - description = f"Upload {upload.id} or item {item_id} not found" - raise FossologyApiError(description, response) - else: - description = f"API error while getting clearing history for item {item_id} from upload {upload.uploadname}." - raise FossologyApiError(description, response) - - def get_prev_next( - self, upload: Upload, item_id: int, selection: PrevNextSelection | None = None - ) -> GetPrevNextItem: - """Get the index of the previous and the next time for an upload - - API Endpoint: GET /uploads/{id}/item/{itemId}/prev-next - - :param upload: the upload to get items from - :param item_id: the id of the item with clearing decision - :param selection: tell Fossology server how to select prev-next item - :type upload: Upload - :type item_id: int - :type selection: str - :return: list of items for the clearing history - :rtype: List[GetPrevNextItem] - :raises FossologyApiError: if the REST call failed - :raises AuthorizationError: if the REST call is not authorized - """ - params = {} - if selection: - params["selection"] = selection - - response = self.session.get( - f"{self.api}/uploads/{upload.id}/item/{item_id}/prev-next", params=params - ) - - if response.status_code == 200: - return GetPrevNextItem.from_json(response.json()) - - elif response.status_code == 404: - description = f"Upload {upload.id} or item {item_id} not found" - raise FossologyApiError(description, response) - else: - description = f"API error while getting prev-next items for {item_id} from upload {upload.uploadname}." - raise FossologyApiError(description, response) - - def get_bulk_history( - self, - upload: Upload, - item_id: int, - ) -> list[GetBulkHistory]: - """Get the bulk history for a specific upload item - - API Endpoint: GET /uploads/{id}/item/{itemId}/bulk-history - - :param upload: the upload to get items from - :param item_id: the id of the item with clearing decision - :type upload: Upload - :type item_id: int - :return: list of data from the bulk history - :rtype: List[GetBulkHistory] - :raises FossologyApiError: if the REST call failed - :raises AuthorizationError: if the REST call is not authorized - """ - response = self.session.get( - f"{self.api}/uploads/{upload.id}/item/{item_id}/bulk-history" - ) - - if response.status_code == 200: - bulk_history = [] - for item in response.json(): - bulk_history.append(GetBulkHistory.from_json(item)) - return bulk_history - - elif response.status_code == 404: - description = f"Upload {upload.id} or item {item_id} not found" - raise FossologyApiError(description, response) - else: - description = f"API error while getting bulk history for {item_id} from upload {upload.uploadname}." - raise FossologyApiError(description, response) - - def schedule_bulk_scan( - self, - upload: Upload, - item_id: int, - spec: dict, - ): - """Schedule a bulk scan for a specific upload item - - API Endpoint: POST /uploads/{id}/item/{itemId}/bulk-scan - - Bulk scan specifications `spec` are added to the request body, - following options are available: - - >>> bulk_scan_spec = { - ... "bulkActions": [ - ... { - ... "licenseShortName": 'MIT', - ... "licenseText": 'License text', - ... "acknowledgement": 'Acknowledgment text', - ... "comment": 'Comment text', - ... "licenseAction": 'ADD', # or 'REMOVE' - ... } - ... ], - ... "refText": 'Reference Text', - ... "bulkScope": 'folder', # or upload - ... "forceDecision": 'false', - ... "ignoreIrre": 'false', - ... "delimiters": 'DEFAULT', - ... "scanOnlyFindings": 'true', - ... } - - :param upload: the upload for the bulk scan - :param item_id: the id of the item for the bulk scan - :param spec: bulk scan specification - :type upload: Upload - :type item_id: int - :raises FossologyApiError: if the REST call failed - :raises AuthorizationError: if the REST call is not authorized - """ - headers = {"Content-Type": "application/json"} - response = self.session.post( - f"{self.api}/uploads/{upload.id}/item/{item_id}/bulk-scan", - headers=headers, - data=json.dumps(spec), - ) - if response.status_code == 201: - logger.info( - f"Bulk scan scheduled for upload {upload.uploadname}, item {item_id}" - ) - elif response.status_code == 400: - description = ( - f"Bad bulk scan request for upload {upload.id}, item {item_id}" - ) - raise FossologyApiError(description, response) - elif response.status_code == 404: - description = f"Upload {upload.id} or item {item_id} not found" - raise FossologyApiError(description, response) - else: - description = f"API error while scheduling bulk scan for item {item_id} from upload {upload.uploadname}." - raise FossologyApiError(description, response) diff --git a/tests/test_upload_clearing.py b/tests/test_items.py similarity index 74% rename from tests/test_upload_clearing.py rename to tests/test_items.py index 1c2e434..aeb8933 100644 --- a/tests/test_upload_clearing.py +++ b/tests/test_items.py @@ -7,7 +7,75 @@ from fossology import Fossology from fossology.enums import PrevNextSelection from fossology.exceptions import FossologyApiError -from fossology.obj import Upload +from fossology.obj import FileInfo, Upload + + +def test_item_info(foss: Fossology, upload_with_jobs: Upload): + files, _ = foss.search(license="BSD") + info: FileInfo = foss.item_info(upload_with_jobs, files[0].uploadTreeId) + print(info.view_info) + print(info.meta_info) + print(info.tag_info) + print(info.package_info) + print(info.reuse_info) + assert info + + +def test_item_info_with_unknown_item_raises_api_error( + foss: Fossology, upload_with_jobs: Upload +): + with pytest.raises(FossologyApiError) as excinfo: + foss.item_info(upload_with_jobs, 1) + assert f"Upload {upload_with_jobs.id} or item 1 not found" in str(excinfo.value) + + +@responses.activate +def test_item_info_500_error( + foss: Fossology, foss_server: str, upload_with_jobs: Upload +): + responses.add( + responses.GET, + f"{foss_server}/api/v1/uploads/{upload_with_jobs.id}/item/1/info", + status=500, + ) + with pytest.raises(FossologyApiError) as excinfo: + foss.item_info(upload_with_jobs, 1) + assert ( + f"API error while getting info for item 1 from upload {upload_with_jobs.uploadname}." + in excinfo.value.message + ) + + +def test_item_copyrights(foss: Fossology, upload_with_jobs: Upload): + files, _ = foss.search(license="BSD") + num_copyrights = foss.item_copyrights(upload_with_jobs, files[0].uploadTreeId) + print(num_copyrights) + assert num_copyrights + + +def test_item_copyrights_with_unknown_item_raises_api_error( + foss: Fossology, upload_with_jobs: Upload +): + with pytest.raises(FossologyApiError) as excinfo: + foss.item_copyrights(upload_with_jobs, 1) + assert f"Upload {upload_with_jobs.id} or item 1 not found" in str(excinfo.value) + + +@responses.activate +def test_item_copyrights_500_error( + foss: Fossology, foss_server: str, upload_with_jobs: Upload +): + responses.add( + responses.GET, + f"{foss_server}/api/v1/uploads/{upload_with_jobs.id}/item/1/totalcopyrights", + status=500, + ) + with pytest.raises(FossologyApiError) as excinfo: + foss.item_copyrights(upload_with_jobs, 1) + assert ( + f"API error while getting copyrights for item 1 from upload {upload_with_jobs.uploadname}." + in excinfo.value.message + ) @pytest.mark.skip(reason="Not yet released, waiting for API version 1.6.0")