feat(uploads): wrap agents and agents/revision endpoints

fossology/obj.py

@@ -527,6 +527,52 @@ def from_json(cls, json_dict):
         return cls(**json_dict)
 
 
+class AgentOfUpload(object):
+    """FOSSology agent that has run on an upload."""
+
+    def __init__(
+        self,
+        uploadId=None,
+        agentName=None,
+        currentAgentId=None,
+        currentAgentRev=None,
+        isAgentRunning=None,
+        successfulAgents=None,
+        **kwargs,
+    ):
+        self.uploadId = uploadId
+        self.agentName = agentName
+        self.currentAgentId = currentAgentId
+        self.currentAgentRev = currentAgentRev
+        self.isAgentRunning = isAgentRunning
+        self.successfulAgents = successfulAgents
+        self.additional_info = kwargs
+
+    def __str__(self):
+        return f"Agent {self.agentName} on upload {self.uploadId}"
+
+    @classmethod
+    def from_json(cls, json_dict):
+        return cls(**json_dict)
+
+
+class AgentsRevision(object):
+    """FOSSology successful agent revision for an upload."""
+
+    def __init__(self, id=None, name=None, revision=None, **kwargs):
+        self.id = id
+        self.name = name
+        self.revision = revision
+        self.additional_info = kwargs
+
+    def __str__(self):
+        return f"Agent {self.name} ({self.id}) revision {self.revision}"
+
+    @classmethod
+    def from_json(cls, json_dict):
+        return cls(**json_dict)
+
+
 class Job(object):
     """FOSSology job."""
 

fossology/uploads.py

@@ -13,6 +13,8 @@
 from fossology.enums import AccessLevel, ClearingStatus
 from fossology.exceptions import AuthorizationError, FossologyApiError
 from fossology.obj import (
+    AgentOfUpload,
+    AgentsRevision,
     Folder,
     Group,
     Permission,
@@ -361,6 +363,66 @@ def upload_summary(self, upload: Upload, group=None):
             description = f"No summary for upload {upload.uploadname} (id={upload.id})"
             raise FossologyApiError(description, response)
 
+    def upload_agents(self, upload: Upload) -> list[AgentOfUpload]:
+        """Get the agents that have run on an upload
+
+        API Endpoint: GET /uploads/{id}/agents
+
+        :param upload: the upload to gather data from
+        :type upload: Upload
+        :return: the list of agents for the upload
+        :rtype: list[AgentOfUpload]
+        :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}/agents"  # type: ignore
+        )
+
+        if response.status_code == 200:
+            return [AgentOfUpload.from_json(agent) for agent in response.json()]
+
+        elif response.status_code == 403:
+            description = f"Getting agents 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 agents for upload {upload.uploadname} (id={upload.id})"
+            raise FossologyApiError(description, response)
+
+    def upload_agents_revision(self, upload: Upload) -> list[AgentsRevision]:
+        """Get the revisions of the successful agents for an upload
+
+        API Endpoint: GET /uploads/{id}/agents/revision
+
+        :param upload: the upload to gather data from
+        :type upload: Upload
+        :return: the list of successful agents with their revision
+        :rtype: list[AgentsRevision]
+        :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}/agents/revision"  # type: ignore
+        )
+
+        if response.status_code == 200:
+            return [AgentsRevision.from_json(agent) for agent in response.json()]
+
+        elif response.status_code == 403:
+            description = (
+                f"Getting agent revisions 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 agent revisions 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,99 @@ def test_upload_summary_with_unknown_group_raises_authorization_error(
     )
 
 
+def test_upload_agents(foss: Fossology, upload_with_jobs: Upload):
+    agents = foss.upload_agents(upload_with_jobs)
+    assert isinstance(agents, list)
+
+
+@responses.activate
+def test_upload_agents_payload(foss: Fossology, foss_server: str, upload: Upload):
+    responses.add(
+        responses.GET,
+        f"{foss_server}/api/v1/uploads/{upload.id}/agents",
+        status=200,
+        json=[
+            {
+                "uploadId": upload.id,
+                "agentName": "nomos",
+                "currentAgentId": 20,
+                "currentAgentRev": "4.1.0",
+                "isAgentRunning": False,
+                "successfulAgents": [
+                    {"agent_id": 20, "agent_rev": "4.1.0", "agent_name": "nomos"}
+                ],
+            }
+        ],
+    )
+    agents = foss.upload_agents(upload)
+    assert len(agents) == 1
+    assert agents[0].agentName == "nomos"
+    assert agents[0].uploadId == upload.id
+    assert str(agents[0]) == f"Agent nomos on upload {upload.id}"
+
+
+@responses.activate
+def test_upload_agents_unauthorized(
+    foss: Fossology, foss_server: str, upload: Upload
+):
+    responses.add(
+        responses.GET,
+        f"{foss_server}/api/v1/uploads/{upload.id}/agents",
+        status=403,
+    )
+    with pytest.raises(AuthorizationError) as excinfo:
+        foss.upload_agents(upload)
+    assert f"Getting agents for upload {upload.id} is not authorized" in str(
+        excinfo.value
+    )
+
+
+@responses.activate
+def test_upload_agents_500_error(foss: Fossology, foss_server: str, upload: Upload):
+    responses.add(
+        responses.GET,
+        f"{foss_server}/api/v1/uploads/{upload.id}/agents",
+        status=500,
+    )
+    with pytest.raises(FossologyApiError):
+        foss.upload_agents(upload)
+
+
+def test_upload_agents_revision(foss: Fossology, upload_with_jobs: Upload):
+    revisions = foss.upload_agents_revision(upload_with_jobs)
+    assert isinstance(revisions, list)
+
+
+@responses.activate
+def test_upload_agents_revision_payload(
+    foss: Fossology, foss_server: str, upload: Upload
+):
+    responses.add(
+        responses.GET,
+        f"{foss_server}/api/v1/uploads/{upload.id}/agents/revision",
+        status=200,
+        json=[{"id": 68, "name": "monk", "revision": "4.1.0.283-rc1"}],
+    )
+    revisions = foss.upload_agents_revision(upload)
+    assert len(revisions) == 1
+    assert revisions[0].name == "monk"
+    assert revisions[0].revision == "4.1.0.283-rc1"
+    assert str(revisions[0]) == "Agent monk (68) revision 4.1.0.283-rc1"
+
+
+@responses.activate
+def test_upload_agents_revision_500_error(
+    foss: Fossology, foss_server: str, upload: Upload
+):
+    responses.add(
+        responses.GET,
+        f"{foss_server}/api/v1/uploads/{upload.id}/agents/revision",
+        status=500,
+    )
+    with pytest.raises(FossologyApiError):
+        foss.upload_agents_revision(upload)
+
+
 def test_delete_if_unknown_upload_raises_error(foss: Fossology, fake_hash: dict):
     upload = Upload(
         foss.rootFolder,