feat: support jobs/history endpoint

fossology/jobs.py

@@ -8,7 +8,7 @@
 from typing import Optional
 
 from fossology.exceptions import AuthorizationError, FossologyApiError
-from fossology.obj import Folder, Job, Upload
+from fossology.obj import Folder, Job, ShowJob, Upload
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
@@ -117,6 +117,31 @@ def detail_job(self, job_id: int, wait: bool = False, timeout: int = 10) -> Job:
             description = f"Error while getting details for job {job_id}"
             raise FossologyApiError(description, response)
 
+    def jobs_history(self, upload: Upload) -> list[Job]:
+        """Return jobs for the given upload id
+
+        API Endpoint: GET /jobs/history
+
+        :param upload_id: the id of the upload
+        :type upload_id: int
+        :return: the list of all the jobs scheduled for the upload
+        :rtype: list of ShowJob
+        :raises FossologyApiError: if the REST call failed
+        """
+        response = self.session.get(f"{self.api}/jobs/history?upload={upload.id}")
+        jobs_list = list()
+        if response.status_code == 200:
+            logger.debug(f"Got jobs for upload {upload.id}")
+            for job in response.json():
+                jobs_list.append(ShowJob.from_json(job))
+            return jobs_list
+        elif response.status_code in [403, 404]:
+            description = f"Upload {upload.id} is not accessible or does not exists"
+            raise FossologyApiError(description, response)
+        else:
+            description = f"Error while getting jobs history for upload {upload.id}"
+            raise FossologyApiError(description, response)
+
     def schedule_jobs(
         self,
         folder: Folder,

fossology/obj.py

@@ -816,7 +816,7 @@ def __init__(
 
     def __str__(self):
         return (
-            f"Job '{self.name}' ({self.id}) queued on {self.queueDate} "
+            f"Job '{self.name}' ({self.id}) queued on {self.queueDate}"
             f"(Status: {self.status} ETA: {self.eta})"
         )
 
@@ -825,6 +825,147 @@ def from_json(cls, json_dict):
         return cls(**json_dict)
 
 
+class JobDownload(object):
+    """FOSSology job download
+
+    Represents a FOSSology job download.
+
+    :param text: text for download link
+    :param link: link to download the report
+    :param kwargs: handle any other job download information provided by the fossology instance
+    :type text: string
+    :type link: string
+    :type kwargs: dict
+    """
+
+    def __init__(self, text: str, link: str, **kwargs):
+        self.text = text
+        self.link = link
+        self.additional_info = kwargs
+
+    def __str__(self):
+        return f"Job output {self.text} can be downloaded here: {self.link}"
+
+    @classmethod
+    def from_json(cls, json_dict):
+        return cls(**json_dict)
+
+
+class JobQueue(object):
+    """FOSSology job queue.
+
+    Represents a FOSSology job queue.
+
+    :param jobQueueId: job queue ID
+    :param jobQueueType: job queue type (agent name)
+    :param startTime: job queue start time
+    :param endTime: job queue end time
+    :param status: job queue complemention status
+    :param itemsProcessed: number of items processes
+    :param log: location of the log file (if it exists)
+    :param dependencies: list of dependent job queue ids
+    :param itemsPerSec: number of items processed per second
+    :param canDoActions: job can accept new actions like pause and cancel
+    :param isInProgress: checks if the job queue is still in progress
+    :param isReady: is the job ready
+    :param download: report download information
+    :param kwargs: handle any other job queue information provided by the fossology instance
+    :type jobQueueId: int
+    :type jobQueueType: string
+    :type startTime: string
+    :type endTime: string
+    :type status: string
+    :type itemsProcessed: int
+    :type log: string
+    :type dependencies: list(int)
+    :type itemsPerSec: float
+    :type canDoActions: bool
+    :type isInProgress: bool
+    :type isReady: bool
+    :type download: JobDownload
+    :type kwargs: key word argument
+    """
+
+    def __init__(
+        self,
+        jobQueueId: int,
+        jobQueueType: str,
+        startTime: str,
+        endTime: str,
+        status: str,
+        itemsProcessed: int,
+        log: str,
+        dependencies: list[int],
+        itemsPerSec: int,
+        canDoActions: bool,
+        isInProgress: bool,
+        isReady: bool,
+        download: JobDownload,
+        **kwargs,
+    ):
+        self.id = jobQueueId
+        self.jobQueueType = jobQueueType
+        self.startTime = startTime
+        self.endTime = endTime
+        self.status = status
+        self.itemsProcessed = itemsProcessed
+        self.log = log
+        self.dependencies = dependencies
+        self.itemsPerSec = itemsPerSec
+        self.canDoActions = canDoActions
+        self.isInProgress = isInProgress
+        self.isReady = isReady
+        self.download = JobDownload.from_json(download) if download else None
+        self.additional_info = kwargs
+
+    def __str__(self):
+        return (
+            f"Job '{self.jobQueueType}' ({self.id}) queued on {self.startTime} processed {self.itemsProcessed} items"
+            f"(Status: {self.status} EndTime: {self.endTime})"
+        )
+
+    @classmethod
+    def from_json(cls, json_dict):
+        return cls(**json_dict)
+
+
+class ShowJob(object):
+    """FOSSology job
+
+    Represents the history of all the jobs and the job queue info
+
+    :param jobId: job ID
+    :param jobName: job name (generally upload name)
+    :param jobQueue: jobs queued for the current job
+    :param uploadId: upload ID to which the job belongs to
+    :type jobId: int
+    :type jobName: string
+    :type jobQueue: list of JobQueue
+    :type uploadId: int
+    """
+
+    def __init__(
+        self,
+        jobId: int,
+        jobName: str,
+        jobQueue: list[JobQueue],
+        uploadId: int,
+    ):
+        self.id = jobId
+        self.jobName = jobName
+        self.jobQueue = list()
+        for job in jobQueue:
+            self.jobQueue.append(JobQueue.from_json(job))
+        self.uploadId = uploadId
+
+    def __str__(self):
+        return f"Job {self.jobName} ({self.id}) for upload {self.uploadId} with {len(self.jobQueue)} jobs in the queue"
+
+    @classmethod
+    def from_json(cls, json_dict):
+        return cls(**json_dict)
+
+
 class ApiLicense(object):
     """FOSSology API License.
 

tests/test_jobs.py

@@ -59,6 +59,19 @@ def test_detail_job_wait_completed(
     mocked_logger.debug.assert_called_once_with((f"Job {job.id} has completed"))
 
 
+def test_jobs_history(
+    foss: Fossology, upload_with_jobs: Upload, foss_schedule_agents: Dict
+):
+    job = foss.schedule_jobs(foss.rootFolder, upload_with_jobs, foss_schedule_agents)
+    assert job.name == upload_with_jobs.uploadname
+
+    jobs = foss.jobs_history(upload=upload_with_jobs)
+    assert len(jobs) == 3
+    assert jobs[0].jobQueue[0].jobQueueType == "reuser"
+    assert jobs[1].jobQueue[0].jobQueueType == "nomos"
+    assert jobs[2].jobQueue[0].jobQueueType == "ununpack"
+
+
 @responses.activate
 def test_schedule_job_error(foss_server: str, foss: Fossology, upload: Upload):
     responses.add(responses.POST, f"{foss_server}/api/v1/jobs", status=404)
@@ -98,6 +111,19 @@ def test_detail_job_error(foss_server: str, foss: Fossology):
     assert f"Error while getting details for job {job_id}" in str(excinfo.value)
 
 
+@responses.activate
+def test_jobs_history_error(foss_server: str, foss: Fossology, fake_hash: str):
+    upload = Upload(secrets.randbelow(1000), "test", secrets.randbelow(1000), "Test upload", "Test name", "24.01.25", hash=fake_hash)
+    responses.add(responses.GET, f"{foss_server}/api/v1/jobs/history", status=403)
+    responses.add(responses.GET, f"{foss_server}/api/v1/jobs/history", status=500)
+    with pytest.raises(FossologyApiError) as excinfo:
+        foss.jobs_history(upload)
+    assert f"Upload {upload.id} is not accessible or does not exists" in str(excinfo.value)
+    with pytest.raises(FossologyApiError) as excinfo:
+        foss.jobs_history(upload)
+    assert f"Error while getting jobs history for upload {upload.id}" in str(excinfo.value)
+
+
 def test_paginated_list_jobs(foss: Fossology, upload_with_jobs: Upload):
     jobs, total_pages = foss.list_jobs(
         upload=upload_with_jobs, page_size=1, all_pages=True