Merge pull request #5772 from fstagni/cherry-pick-2-734faf191-integration

[sweep:integration] DIRAC APIs check if a JobState transition is allowed

Author: fstagni

src/DIRAC/Interfaces/API/Dirac.py

@@ -1675,6 +1675,10 @@ def deleteJob(self, jobID):
             return ret
         jobID = ret["Value"]
 
+        res = JobStatus.checkJobStateTransition(jobID, JobStatus.DELETED)
+        if not res["OK"]:
+            return res
+
         result = WMSClient(useCertificates=self.useCertificates).deleteJob(jobID)
         if result["OK"]:
             if self.jobRepo:
@@ -1705,6 +1709,10 @@ def rescheduleJob(self, jobID):
             return ret
         jobID = ret["Value"]
 
+        res = JobStatus.checkJobStateTransition(jobID, JobStatus.RESCHEDULED)
+        if not res["OK"]:
+            return res
+
         result = WMSClient(useCertificates=self.useCertificates).rescheduleJob(jobID)
         if result["OK"]:
             if self.jobRepo:
@@ -1721,11 +1729,11 @@ def killJob(self, jobID):
 
         Example Usage:
 
-         >>> print dirac.killJob(12345)
+         >>> print(dirac.killJob(12345))
          {'OK': True, 'Value': [12345]}
 
         :param jobID: JobID
-        :type jobID: int, str or python:list
+        :type jobID: int, str
         :returns: S_OK,S_ERROR
 
         """
@@ -1734,6 +1742,10 @@ def killJob(self, jobID):
             return ret
         jobID = ret["Value"]
 
+        res = JobStatus.checkJobStateTransition(jobID, JobStatus.KILLED)
+        if not res["OK"]:
+            return res
+
         result = WMSClient(useCertificates=self.useCertificates).killJob(jobID)
         if result["OK"]:
             if self.jobRepo:

src/DIRAC/WorkloadManagementSystem/Client/JobStatus.py

@@ -1,6 +1,8 @@
 """
 This module contains constants and lists for the possible job states.
 """
+
+from DIRAC import gLogger, S_OK, S_ERROR
 from DIRAC.Core.Utilities.StateMachine import State, StateMachine
 
 
@@ -89,3 +91,33 @@ def __init__(self, state):
             RECEIVED: State(1, [CHECKING, WAITING, FAILED, DELETED], defState=RECEIVED),
             SUBMITTING: State(0, [RECEIVED, CHECKING, DELETED], defState=SUBMITTING),  # initial state
         }
+
+
+def checkJobStateTransition(jobID, candidateState, currentStatus=None, jobMonitoringClient=None):
+    """Utility to check if a job state transition is allowed"""
+    if not currentStatus:
+        if not jobMonitoringClient:
+            from DIRAC.WorkloadManagementSystem.Client.JobMonitoringClient import JobMonitoringClient
+
+            jobMonitoringClient = JobMonitoringClient()
+
+        res = jobMonitoringClient.getJobsStatus(jobID)
+        if not res["OK"]:
+            return res
+        try:
+            currentStatus = res["Value"][jobID]["Status"]
+        except KeyError:
+            return S_ERROR("Job does not exist")
+
+    res = JobsStateMachine(currentStatus).getNextState(candidateState)
+    if not res["OK"]:
+        return res
+
+    # If the JobsStateMachine does not accept the candidate, return an ERROR
+    if candidateState != res["Value"]:
+        gLogger.error(
+            "Job Status Error",
+            "%s can't move from %s to %s" % (jobID, currentStatus, candidateState),
+        )
+        return S_ERROR("Job state transition not allowed")
+    return S_OK()

src/DIRAC/WorkloadManagementSystem/DB/JobDB.py

@@ -650,7 +650,7 @@ def setJobsMajorStatus(self, jIDList, candidateStatus, force=False):
                     return res
                 nextState = res["Value"]
 
-                # If the JobsStateMachine does not accept the candidate, add it to separate dictionary
+                # The JobsStateMachine might force a different status
                 if candidateStatus != nextState:
                     self.log.error(
                         "Job Status Error",