diff --git a/fossology/obj.py b/fossology/obj.py index 89740f2..8d5de0d 100644 --- a/fossology/obj.py +++ b/fossology/obj.py @@ -527,6 +527,45 @@ def from_json(cls, json_dict): return cls(**json_dict) +class ClearingProgress(object): + """FOSSology upload clearing progress.""" + + def __init__(self, totalFilesOfInterest=None, totalFilesCleared=None, **kwargs): + self.totalFilesOfInterest = totalFilesOfInterest + self.totalFilesCleared = totalFilesCleared + self.additional_info = kwargs + + def __str__(self): + return f"{self.totalFilesCleared}/{self.totalFilesOfInterest} files cleared" + + @classmethod + def from_json(cls, json_dict): + return cls(**json_dict) + + +class LicenseHistogram(object): + """FOSSology license histogram entry for an upload.""" + + def __init__( + self, id=None, name=None, scannerCount=None, concludedCount=None, **kwargs + ): + self.id = id + self.name = name + self.scannerCount = scannerCount + self.concludedCount = concludedCount + self.additional_info = kwargs + + def __str__(self): + return ( + f"License {self.name} ({self.id}): {self.scannerCount} scanned, " + f"{self.concludedCount} concluded" + ) + + @classmethod + def from_json(cls, json_dict): + return cls(**json_dict) + + class Job(object): """FOSSology job.""" diff --git a/fossology/uploads.py b/fossology/uploads.py index 82e9a41..bf03c98 100644 --- a/fossology/uploads.py +++ b/fossology/uploads.py @@ -13,8 +13,10 @@ from fossology.enums import AccessLevel, ClearingStatus from fossology.exceptions import AuthorizationError, FossologyApiError from fossology.obj import ( + ClearingProgress, Folder, Group, + LicenseHistogram, Permission, Summary, Upload, @@ -361,6 +363,76 @@ def upload_summary(self, upload: Upload, group=None): description = f"No summary for upload {upload.uploadname} (id={upload.id})" raise FossologyApiError(description, response) + def clearing_progress(self, upload: Upload) -> ClearingProgress: + """Get the clearing progress information for an upload + + API Endpoint: GET /uploads/{id}/clearing-progress + + :param upload: the upload to gather data from + :type upload: Upload + :return: the clearing progress of the upload + :rtype: ClearingProgress + :raises FossologyApiError: if the REST call failed + :raises AuthorizationError: if the REST call is not authorized + """ + response = self.session.get( # type: ignore + f"{self.api}/uploads/{upload.id}/clearing-progress" # type: ignore + ) + + if response.status_code == 200: + return ClearingProgress.from_json(response.json()) + + elif response.status_code == 403: + description = ( + f"Getting clearing progress for upload {upload.id} is not authorized" + ) + raise AuthorizationError(description, response) + elif response.status_code == 404: + description = f"Upload {upload.id} not found" + raise FossologyApiError(description, response) + else: + description = f"Unable to get clearing progress for upload {upload.uploadname} (id={upload.id})" + raise FossologyApiError(description, response) + + def license_histogram( + self, upload: Upload, agent_id: int | None = None + ) -> list[LicenseHistogram]: + """Get the licenses histogram for an upload + + API Endpoint: GET /uploads/{id}/licenses/histogram + + :param upload: the upload to gather data from + :param agent_id: limit the histogram to a specific agent (default: None) + :type upload: Upload + :type agent_id: int | None + :return: the license histogram of the upload + :rtype: list[LicenseHistogram] + :raises FossologyApiError: if the REST call failed + :raises AuthorizationError: if the REST call is not authorized + """ + params = {} + if agent_id is not None: + params["agentId"] = agent_id + response = self.session.get( # type: ignore + f"{self.api}/uploads/{upload.id}/licenses/histogram", # type: ignore + params=params, # type: ignore + ) + + if response.status_code == 200: + return [LicenseHistogram.from_json(item) for item in response.json()] + + elif response.status_code == 403: + description = ( + f"Getting license histogram for upload {upload.id} is not authorized" + ) + raise AuthorizationError(description, response) + elif response.status_code == 404: + description = f"Upload {upload.id} not found" + raise FossologyApiError(description, response) + else: + description = f"Unable to get license histogram for upload {upload.uploadname} (id={upload.id})" + raise FossologyApiError(description, response) + @retry(retry=retry_if_exception_type(TryAgain), stop=stop_after_attempt(3)) def upload_licenses( self, diff --git a/tests/test_uploads.py b/tests/test_uploads.py index 79b245c..c79914e 100644 --- a/tests/test_uploads.py +++ b/tests/test_uploads.py @@ -326,6 +326,107 @@ def test_upload_summary_with_unknown_group_raises_authorization_error( ) +def test_clearing_progress(foss: Fossology, upload: Upload): + progress = foss.clearing_progress(upload) + assert progress.totalFilesCleared is not None + assert progress.totalFilesOfInterest is not None + assert "files cleared" in str(progress) + + +@responses.activate +def test_clearing_progress_not_found( + foss: Fossology, foss_server: str, upload: Upload +): + responses.add( + responses.GET, + f"{foss_server}/api/v1/uploads/{upload.id}/clearing-progress", + status=404, + ) + with pytest.raises(FossologyApiError) as excinfo: + foss.clearing_progress(upload) + assert f"Upload {upload.id} not found" in str(excinfo.value) + + +@responses.activate +def test_clearing_progress_unauthorized( + foss: Fossology, foss_server: str, upload: Upload +): + responses.add( + responses.GET, + f"{foss_server}/api/v1/uploads/{upload.id}/clearing-progress", + status=403, + ) + with pytest.raises(AuthorizationError) as excinfo: + foss.clearing_progress(upload) + assert f"Getting clearing progress for upload {upload.id} is not authorized" in str( + excinfo.value + ) + + +@responses.activate +def test_clearing_progress_500_error( + foss: Fossology, foss_server: str, upload: Upload +): + responses.add( + responses.GET, + f"{foss_server}/api/v1/uploads/{upload.id}/clearing-progress", + status=500, + ) + with pytest.raises(FossologyApiError): + foss.clearing_progress(upload) + + +def test_license_histogram(foss: Fossology, upload_with_jobs: Upload): + histogram = foss.license_histogram(upload_with_jobs) + assert isinstance(histogram, list) + + +@responses.activate +def test_license_histogram_payload( + foss: Fossology, foss_server: str, upload: Upload +): + responses.add( + responses.GET, + f"{foss_server}/api/v1/uploads/{upload.id}/licenses/histogram", + status=200, + json=[{"id": 357, "name": "MIT", "scannerCount": 2, "concludedCount": 1}], + ) + histogram = foss.license_histogram(upload, agent_id=3) + assert len(histogram) == 1 + assert histogram[0].name == "MIT" + assert histogram[0].scannerCount == 2 + assert "agentId=3" in responses.calls[0].request.url + + +@responses.activate +def test_license_histogram_unauthorized( + foss: Fossology, foss_server: str, upload: Upload +): + responses.add( + responses.GET, + f"{foss_server}/api/v1/uploads/{upload.id}/licenses/histogram", + status=403, + ) + with pytest.raises(AuthorizationError) as excinfo: + foss.license_histogram(upload) + assert f"Getting license histogram for upload {upload.id} is not authorized" in str( + excinfo.value + ) + + +@responses.activate +def test_license_histogram_500_error( + foss: Fossology, foss_server: str, upload: Upload +): + responses.add( + responses.GET, + f"{foss_server}/api/v1/uploads/{upload.id}/licenses/histogram", + status=500, + ) + with pytest.raises(FossologyApiError): + foss.license_histogram(upload) + + def test_delete_if_unknown_upload_raises_error(foss: Fossology, fake_hash: dict): upload = Upload( foss.rootFolder,