feat: Improved JobMonitoring and rebase on top of legacy adaptor

Author: Robin-Van-de-Merghel

.github/workflows/integration.yml

@@ -1,8 +1,8 @@
 name: Integration tests
 
 env:
-  DIRACX_REPO: default
-  DIRACX_BRANCH: main
+  DIRACX_REPO: https://github.com/Robin-Van-de-Merghel/diracx.git
+  DIRACX_BRANCH: robin-update-job-monitoring
 
 on:
   push:

src/DIRAC/WorkloadManagementSystem/Client/JobMonitoringClient.py

@@ -18,8 +18,7 @@ def __init__(self, **kwargs):
         super().__init__(**kwargs)
         self.setServer("WorkloadManagement/JobMonitoring")
 
-    # Set to NULL to avoid using it in ClientSelector
-    diracxClient = None
+    diracxClient = futureJobMonitoringClient
 
     @ignoreEncodeWarning
     def getJobsStatus(self, jobIDs):

src/DIRAC/WorkloadManagementSystem/FutureClient/JobMonitoringClient.py

@@ -5,6 +5,8 @@
 
 class JobMonitoringClient:
     def fetch(self, parameters, jobIDs):
+        if not isinstance(jobIDs, list):
+            jobIDs = [jobIDs]
         with DiracXClient() as api:
             jobs = api.jobs.search(
                 parameters=["JobID"] + parameters,
@@ -34,12 +36,10 @@ def getJobsStatus(self, jobIDs):
 
     @convertToReturnValue
     def getJobParameter(self, jobID, parName):
-        # with DiracXClient() as api:
-        #     parameters = api.jobs.get_job_parameters(
-        #         job_id=jobID
-        #     )
+        with DiracXClient() as api:
+            parameters = api.jobs.get_job_parameters(job_id=jobID)
 
-        #     return parameters
+            return parameters
         pass
 
     @convertToReturnValue
@@ -94,11 +94,55 @@ def cleanLFN(data):
 
             return return_dict
 
+    def convert_condDict_to_searchSpec(self, condDict):
+        """
+        Convert condDict to list of SearchSpec dicts.
+
+        - If key is str: one SearchSpec
+        - If key is tuple: multiple SearchSpec (one per key in tuple)
+        - If value is dict: skip this condDict entry
+        - list value -> operator 'in'
+        - scalar value -> operator 'eq'
+        """
+        search_list = []
+
+        if not condDict:
+            return search_list
+
+        for key, val in condDict.items():
+            if isinstance(val, dict):
+                # abandon this condition entry
+                continue
+
+            # Normalize keys: if tuple, treat each element as a separate key
+            keys = [key] if isinstance(key, str) else list(key)
+
+            for k in keys:
+                if not isinstance(k, str):
+                    # skip non-str keys silently (or raise error if you want)
+                    continue
+
+                if isinstance(val, list):
+                    search_list.append({"parameter": k, "operator": "in", "values": val})
+                else:
+                    search_list.append({"parameter": k, "operator": "eq", "value": val})
+
+        return search_list
+
     @convertToReturnValue
-    def getJobs(self, attrDict, cutDate=None):
-        # TODO: Check, selectJobs is awful...
-        # See if we do it
-        pass
+    def getJobs(self, attrDict={}, cutDate=None):
+        # Get job >>>IDS<<<
+        with DiracXClient() as api:
+            search = self.convert_condDict_to_searchSpec(attrDict)
+
+            if cutDate:
+                # TODO: Verify the date format!
+                search.append({"parameter": "LastUpdateTime", "operator": "gt", "value": str(cutDate)})
+
+            jobs = api.jobs.search(parameters=["JobID"], search=search)
+
+            # Returns a **string**??, cf test_JobStateUpdateAndJobMonitoringMultiple
+            return [str(job["JobID"]) for job in jobs]
 
     @convertToReturnValue
     def getJobJDL(self, jobID, original):
@@ -133,7 +177,15 @@ def getJobLoggingInfo(self, jobID):
 
     @convertToReturnValue
     def getJobSummary(self, jobID):
-        return self.fetch(parameters=[], jobIDs=[jobID])
+        with DiracXClient() as api:
+            # TODO: Fix diracx client...
+            jobs = api.jobs.summary(
+                grouping=[], search=[{"parameter": "JobID", "operator": "eq", "value": jobID}]  # type: ignore
+            )
+
+            if jobs:
+                return jobs[0]
+            return []
 
     @convertToReturnValue
     def getJobsSummary(self, jobIDs):