diff --git a/fossology/obj.py b/fossology/obj.py index 89740f2..cef1f1b 100644 --- a/fossology/obj.py +++ b/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.""" diff --git a/fossology/uploads.py b/fossology/uploads.py index 82e9a41..397f6dd 100644 --- a/fossology/uploads.py +++ b/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, diff --git a/tests/test_uploads.py b/tests/test_uploads.py index 79b245c..0fbbcb8 100644 --- a/tests/test_uploads.py +++ b/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,