feat(uploads): wrap clearing-progress and licenses/histogram endpoints

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."""
 

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,

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,